fix: route remote control through provider startup args

Gitea returns HTTP 422 when a deploy key title or public key content
already exists on the repo. The provisioner previously surfaced this
as a generic RuntimeError with the raw status code. Introduce
DeployKeyCollisionError (a RuntimeError subclass) in the base module
and detect 422 in GiteaDeployKeyProvisioner.create so callers can
catch collisions explicitly and the error message names the repo and
title involved.

README.md

@@ -14,7 +14,8 @@
 
 ## Features
 
-- **Per-bottle egress allowlist** — TLS-bumped HTTP/HTTPS chokepoint with a per-manifest host allowlist and request-body DLP scanner; DoH and arbitrary hosts blocked by default.
+- **Per-bottle egress allowlist** — TLS-bumped HTTP/HTTPS chokepoint with a per-manifest host allowlist; per-route path/method/header `matches` filtering; outbound DLP scanning for known tokens and secrets, inbound DLP scanning for prompt-injection attempts; DoH and arbitrary hosts blocked by default.
+- **Per-route token-match policy** — each egress route picks what happens when the outbound DLP catches a token via `dlp.outbound_on_match`: `supervise` (default) holds the request and surfaces it in `./cli.py supervise` for approval (an approved value is remembered for the life of the proxy); `redact` scrubs the value and forwards; `block` is a hard `403`. Cuts false-positive friction without weakening default-deny.
 - **Tokens the agent never sees** — host secrets live in a sidecar; the agent dials `http://sidecar:9099/<path>` and the proxy strips inbound `Authorization` and injects the real token before forwarding. `printenv` in the agent shows proxy URLs only.
 - **Gitleaks-scanned push (git-gate)** — `bottle.git` remotes route through a per-bottle `git daemon` that gitleaks-scans incoming refs pre-receive and forwards clean refs upstream over SSH. The agent never holds the upstream credential.
 - **Manifest-scoped skills + secrets** — each bottle declares its skills, env, git identity, remotes, and egress routes; unknown keys die at load.
@@ -106,8 +107,15 @@ egress:
   routes:
     - host: gitea.dideric.is
       auth:
-        scheme: token
+        scheme: token        # Bearer | token
         token_ref: BOT_BOTTLE_GITEA_TOKEN
+      matches:               # optional — restrict to specific paths/methods/headers
+        - paths:
+            - {type: prefix, value: /api/v1/}
+          methods: [GET, POST, PATCH, DELETE]
+      dlp:                   # optional — per-route detector overrides (default: all on)
+        outbound_detectors: [token_patterns, known_secrets]
+        inbound_detectors: false   # disable response scanning for this host
 ---
 
 The `gitea-dev` bottle. Provider auth via the inherited Claude route;
@@ -126,6 +134,26 @@ skills:
 You help maintain Gitea-hosted projects.
 ````
 
+**Egress route fields:**
+
+| Field | Required | Description |
+|---|---|---|
+| `host` | yes | Hostname to allowlist. One entry per host. |
+| `role` | no | Reserved for future use. The key is recognised but any value is currently rejected at load. Provider auth routes (e.g. Claude's `api.anthropic.com`) are injected automatically from `agent_provider.auth_token`, not via `role`. |
+| `auth.scheme` | when `auth` present | `Bearer` or `token`. Injected by the proxy; the agent never sees the value. |
+| `auth.token_ref` | when `auth` present | Env-var name holding the secret on the host. |
+| `matches` | no | Array of `{paths, methods, headers}` filters. A request must match at least one entry (if any are given) to be forwarded. |
+| `matches[].paths` | no | Array of `{type, value}`. `type` is `prefix` (default), `exact`, or `regex`. |
+| `matches[].methods` | no | Array of HTTP method strings, e.g. `[GET, POST]`. |
+| `matches[].headers` | no | Array of `{name, value, type}`. `type` is `exact` (default) or `regex`. |
+| `dlp` | no | Per-route DLP overrides. Omit to use defaults (all detectors on). |
+| `dlp.outbound_detectors` | no | `false` disables outbound scanning; list restricts to named detectors (`token_patterns`, `known_secrets`). |
+| `dlp.inbound_detectors` | no | `false` disables inbound scanning; list restricts to named detectors (`naive_injection_detection`). |
+| `dlp.outbound_on_match` | no | What to do when an outbound token is detected: `supervise` (default for manifest routes — hold for operator approval), `redact` (scrub the value and forward), or `block` (hard 403). Agent-provider routes (e.g. `api.anthropic.com`) default to `redact`. |
+| `git.fetch` | no | `true` permits smart HTTP clone/fetch (`git-upload-pack`) for this host. Push (`git-receive-pack`) remains blocked. |
+
+When an outbound DLP detector matches a token, the route's `dlp.outbound_on_match` policy decides what happens. Under the default `supervise`, the proxy queues an `egress-token-allow` proposal for the operator's `./cli.py supervise` TUI and holds the request open until it is answered (or `EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS`, default 300s, elapses — after which it fails closed). The operator never sees the raw token, only the host, method, path, and a redacted snippet; approving adds the value to an in-memory safelist for the life of the egress proxy. Under `redact`, the matched value is scrubbed from the body, headers, and path and the request is forwarded (failing closed if a match lands somewhere unredactable, like the hostname). Under `block` it stays a hard `403`. Structural blocks (CRLF injection) and not-in-allowlist host blocks are always hard `403`s regardless of policy.
+
 More examples in `examples/`. Full design lives under `docs/prds/`; the trust-boundary rationale is in `docs/prds/0011-per-file-md-manifest.md`.
 
 ## Trademarks

bot_bottle/agent_provider.py

@@ -61,7 +61,6 @@ class AgentProviderRuntime:
     prompt_mode: PromptMode
     bypass_args: tuple[str, ...]
     resume_args: tuple[str, ...]
-    remote_control_args: tuple[str, ...]
 
 
 @dataclass(frozen=True)
@@ -371,6 +370,15 @@ def build_agent_provision_plan(
     )
 
 
+def provider_startup_args(
+    provider_settings: dict[str, object] | None,
+) -> tuple[str, ...]:
+    raw = (provider_settings or {}).get("startup_args", ())
+    if not isinstance(raw, (list, tuple)):
+        return ()
+    return tuple(arg for arg in raw if isinstance(arg, str))
+
+
 def prompt_args(
     prompt_mode: PromptMode,
     prompt_path: str | None,
@@ -382,7 +390,7 @@ def prompt_args(
     if prompt_mode == "append_file":
         return ["--append-system-prompt-file", prompt_path]
     if prompt_mode == "read_prompt_file":
-        if argv and "resume" in argv:
+        if argv and ("resume" in argv or "remote-control" in argv):
             return []
         return [f"Read and follow the instructions in {prompt_path}."]
     if prompt_mode == "print_read_prompt_file":

bot_bottle/backend/__init__.py

@@ -109,9 +109,8 @@ class BottlePlan(ABC):
     def workspace_plan(self) -> WorkspacePlan:
         return workspace_plan(self.spec, guest_home=self.guest_home)
 
-    def print(self, *, remote_control: bool) -> None:
+    def print(self) -> None:
         """Render the y/N preflight summary to stderr."""
-        del remote_control
         spec = self.spec
         manifest = self.manifest
         agent = manifest.agent

bot_bottle/cli/resume.py

@@ -28,7 +28,6 @@ from .start import _launch_bottle
 def cmd_resume(argv: list[str]) -> int:
     parser = argparse.ArgumentParser(prog=f"{PROG} resume", add_help=True)
     parser.add_argument("--dry-run", action="store_true")
-    parser.add_argument("--remote-control", action="store_true")
     parser.add_argument(
         "identity",
         help="bottle identity from a prior `start` (see its session-end output)",
@@ -56,6 +55,5 @@ def cmd_resume(argv: list[str]) -> int:
     return _launch_bottle(
         spec,
         dry_run=args.dry_run,
-        remote_control=args.remote_control,
         backend_name=backend_name,
     )

bot_bottle/cli/start.py

@@ -42,7 +42,6 @@ def cmd_start(argv: list[str]) -> int:
     parser = argparse.ArgumentParser(prog=f"{PROG} start", add_help=True)
     parser.add_argument("--dry-run", action="store_true")
     parser.add_argument("--cwd", action="store_true", help="copy host cwd into the running bottle")
-    parser.add_argument("--remote-control", action="store_true")
     parser.add_argument(
         "--backend",
         choices=known_backend_names(),
@@ -89,7 +88,6 @@ def cmd_start(argv: list[str]) -> int:
     return _launch_bottle(
         spec,
         dry_run=dry_run,
-        remote_control=args.remote_control,
         backend_name=backend_name,
     )
 
@@ -134,7 +132,7 @@ def prepare_with_preflight(
 
 
 def attach_agent(
-    bottle: Bottle, *, remote_control: bool = False, resume: bool = False,
+    bottle: Bottle, *, resume: bool = False,
     agent_provider_template: str = "claude",
     startup_args: tuple[str, ...] = (),
 ) -> int:
@@ -153,8 +151,6 @@ def attach_agent(
         "(Ctrl-D or 'exit' to leave; container will be removed)"
     )
     agent_args = list(runtime.bypass_args)
-    if remote_control:
-        agent_args.extend(runtime.remote_control_args)
     agent_args.extend(startup_args)
     if resume:
         agent_args.extend(runtime.resume_args)
@@ -218,9 +214,9 @@ def _text_prompt_yes() -> bool:
     return reply in ("y", "Y", "yes", "YES")
 
 
-def _text_render_preflight(*, remote_control: bool):
+def _text_render_preflight():
     def _render(plan: DockerBottlePlan) -> None:
-        plan.print(remote_control=remote_control)
+        plan.print()
     return _render
 
 
@@ -228,7 +224,6 @@ def _launch_bottle(
     spec: BottleSpec,
     *,
     dry_run: bool,
-    remote_control: bool,
     backend_name: str | None = None,
 ) -> int:
     """Shared launch core for `start` and `resume`. Builds the plan,
@@ -240,7 +235,7 @@ def _launch_bottle(
         plan, identity = prepare_with_preflight(
             spec,
             stage_dir=stage_dir,
-            render_preflight=_text_render_preflight(remote_control=remote_control),
+            render_preflight=_text_render_preflight(),
             prompt_yes=_text_prompt_yes,
             dry_run=dry_run,
             backend_name=backend_name,
@@ -253,7 +248,6 @@ def _launch_bottle(
             agent_provider_template = getattr(plan, "agent_provider_template", "claude")
             exit_code = attach_agent(
                 bottle,
-                remote_control=remote_control,
                 agent_provider_template=agent_provider_template,
                 startup_args=plan.agent_provision.startup_args,
             )

bot_bottle/cli/supervise.py

@@ -51,8 +51,10 @@ from ..supervise import (
     STATUS_MODIFIED,
     STATUS_REJECTED,
     TOOL_CAPABILITY_BLOCK,
-    TOOL_ALLOW,
+    TOOL_EGRESS_ALLOW,
     TOOL_EGRESS_BLOCK,
+    TOOL_GITLEAKS_ALLOW,
+    TOOL_EGRESS_TOKEN_ALLOW,
     archive_proposal,
     list_pending_proposals,
     render_diff,
@@ -64,6 +66,11 @@ from ._common import PROG
 
 _REFRESH_INTERVAL_MS = 1000
 
+# Proposal tools whose payload is a read-only report, not a file the operator
+# edits: modify is unavailable and approval requires a recorded reason for the
+# audit trail.
+_REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW)
+
 
 @dataclass(frozen=True)
 class QueuedProposal:
@@ -138,8 +145,10 @@ def _detail_lines(
 def _suffix_for_tool(tool: str) -> str:
     if tool == TOOL_CAPABILITY_BLOCK:
         return ".dockerfile"
-    if tool in (TOOL_ALLOW, TOOL_EGRESS_BLOCK):
+    if tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         return ".yaml"
+    if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
+        return ".txt"
     return ".txt"
 
 
@@ -168,7 +177,7 @@ def approve(
     #     diff_before, diff_after = apply_capability_change(
     #         qp.proposal.bottle_slug, file_to_apply,
     #     )
-    if qp.proposal.tool in (TOOL_ALLOW, TOOL_EGRESS_BLOCK):
+    if qp.proposal.tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
         diff_before, diff_after = apply_routes_change(
             qp.proposal.bottle_slug,
             file_to_apply,
@@ -201,6 +210,23 @@ def reject(qp: QueuedProposal, *, reason: str) -> None:
     _write_audit(qp, action=STATUS_REJECTED, notes=reason, diff_before="", diff_after="")
 
 
+def _approve_from_tui(
+    stdscr: "curses._CursesWindow",  # type: ignore
+    qp: QueuedProposal,
+    *,
+    final_file: str | None = None,
+    notes: str = "",
+) -> str:
+    """Approve from curses, prompting for any tool-specific audit note."""
+    if qp.proposal.tool in _REPORT_ONLY_TOOLS and final_file is None:
+        notes = _prompt(stdscr, "allow reason (false positive / legitimately needed): ")
+        if not notes:
+            return "approve aborted (empty reason)"
+    approve(qp, final_file=final_file, notes=notes)
+    verb = "modified+approved" if final_file is not None else "approved"
+    return _approval_status(qp, verb)
+
+
 def _write_audit(
     qp: QueuedProposal,
     *,
@@ -272,7 +298,10 @@ def cmd_supervise(argv: list[str]) -> int:
         return e.code if isinstance(e.code, int) else 1
     except Exception as e:  # noqa: W0718 — catch supervise crash for logging
         log_path = _write_crash_log(e)
-        error(f"supervise crashed: {type(e).__name__}: {e}")
+        error(
+            f"supervise crashed: {type(e).__name__}: {e}",
+            context={"error_type": type(e).__name__, "crash_log": str(log_path)},
+        )
         error(f"full traceback written to {log_path}")
         return 1
     return 0
@@ -384,18 +413,22 @@ def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore
             _detail_view(stdscr, qp, green_attr=green_attr)
         elif key == ord("a"):
             try:
-                approve(qp)
+                status_line = _approve_from_tui(stdscr, qp)
-                status_line = _approval_status(qp, "approved")
             except ApplyError as e:
                 status_line = f"apply failed: {e}"
         elif key == ord("m"):
+            if qp.proposal.tool in _REPORT_ONLY_TOOLS:
+                status_line = f"modify unavailable for {qp.proposal.tool}"
+                continue
             edited = _modify(stdscr, qp)
             if edited is None:
                 status_line = "modify aborted (no change)"
             else:
                 try:
-                    approve(qp, final_file=edited, notes="operator modified before approving")
+                    status_line = _approve_from_tui(
-                    status_line = _approval_status(qp, "modified+approved")
+                        stdscr, qp, final_file=edited,
+                        notes="operator modified before approving",
+                    )
                 except ApplyError as e:
                     status_line = f"apply failed: {e}"
         elif key == ord("r"):
@@ -493,15 +526,20 @@ def _detail_view(
             offset = max(0, len(lines) - 1)
         elif key == ord("a"):
             try:
-                approve(qp)
+                _approve_from_tui(stdscr, qp)
             except ApplyError:
                 pass
             return
         elif key == ord("m"):
+            if qp.proposal.tool in _REPORT_ONLY_TOOLS:
+                return
             edited = _modify(stdscr, qp)
             if edited is not None:
                 try:
-                    approve(qp, final_file=edited, notes="operator modified before approving")
+                    _approve_from_tui(
+                        stdscr, qp, final_file=edited,
+                        notes="operator modified before approving",
+                    )
                 except ApplyError:
                     pass
             return

bot_bottle/contrib/claude/agent_provider.py

@@ -20,6 +20,7 @@ from ...agent_provider import (
     AgentProvisionDir,
     AgentProvisionFile,
     AgentProvisionPlan,
+    provider_startup_args,
 )
 from ...backend.docker import util as docker_mod
 from ...egress import EgressRoute
@@ -90,7 +91,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="append_file",
     bypass_args=("--dangerously-skip-permissions",),
     resume_args=("--continue",),
-    remote_control_args=("--remote-control",),
 )
 
 
@@ -115,8 +115,9 @@ class ClaudeAgentProvider(AgentProvider):
         color: str = "",
         provider_settings: dict[str, object] | None = None,
     ) -> AgentProvisionPlan:
-        del forward_host_credentials, host_env, provider_settings
+        del forward_host_credentials, host_env
         resolved_guest_env = dict(guest_env or {})
+        startup_args = provider_startup_args(provider_settings)
         guest_home = self.guest_home
         trusted_path = trusted_project_path or guest_home
 
@@ -199,6 +200,7 @@ class ClaudeAgentProvider(AgentProvider):
             env_vars=env_vars,
             guest_env=resolved_guest_env,
             has_prompt=has_prompt,
+            startup_args=startup_args,
             dirs=dirs,
             files=tuple(files),
             egress_routes=egress_routes,

bot_bottle/contrib/codex/Dockerfile

@@ -1,12 +1,12 @@
 # bot-bottle Codex provider image.
 #
 # Mirrors the default Claude image shape: Node LTS, git/network tooling,
-# non-root node user, and the provider CLI installed globally.
+# non-root node user, and the provider CLI installed for that user.
 
 FROM node:22-slim
 
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl \
+  && apt-get install -y --no-install-recommends git ca-certificates curl procps \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by codex itself
@@ -17,12 +17,15 @@ RUN apt-get update \
   && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
   && rm -rf /var/lib/apt/lists/*
 
-RUN npm install -g --no-fund --no-audit @openai/codex@0.136.0 \
-  && npm cache clean --force
-
 USER node
 WORKDIR /home/node
 
-RUN mkdir -p /home/node/.codex
+ENV PATH="/home/node/.local/bin:${PATH}"
+
+# Remote-control support requires the standalone Codex install layout
+# under ~/.codex/packages/standalone/current. The npm package can run
+# the TUI, but remote-control commands expect this installer-owned path.
+RUN mkdir -p /home/node/.codex \
+  && curl -fsSL https://chatgpt.com/codex/install.sh | sh
 
 CMD ["codex"]

bot_bottle/contrib/codex/agent_provider.py

@@ -22,6 +22,7 @@ from ...agent_provider import (
     AgentProvisionCommand,
     AgentProvisionFile,
     AgentProvisionPlan,
+    provider_startup_args,
 )
 from .codex_auth import codex_host_access_token, write_codex_dummy_auth_file
 from ...egress import CODEX_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
@@ -54,7 +55,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="read_prompt_file",
     bypass_args=("--dangerously-bypass-approvals-and-sandbox",),
     resume_args=("resume", "--last"),
-    remote_control_args=(),
 )
 
 
@@ -79,8 +79,9 @@ class CodexAgentProvider(AgentProvider):
         color: str = "",
         provider_settings: dict[str, object] | None = None,
     ) -> AgentProvisionPlan:
-        del auth_token, label, color, provider_settings
+        del auth_token, label, color
         resolved_guest_env = dict(guest_env or {})
+        startup_args = provider_startup_args(provider_settings)
         guest_home = self.guest_home
         trusted_path = trusted_project_path or guest_home
 
@@ -163,6 +164,7 @@ class CodexAgentProvider(AgentProvider):
             env_vars=env_vars,
             guest_env=resolved_guest_env,
             has_prompt=has_prompt,
+            startup_args=startup_args,
             dirs=tuple(dirs),
             files=tuple(files),
             pre_copy=tuple(pre_copy),

bot_bottle/contrib/gitea/deploy_key_provisioner.py

@@ -19,7 +19,7 @@ import urllib.error
 import urllib.request
 from pathlib import Path
 
-from ...deploy_key_provisioner import DeployKeyProvisioner
+from ...deploy_key_provisioner import DeployKeyCollisionError, DeployKeyProvisioner
 
 
 class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
@@ -71,6 +71,11 @@ class GiteaDeployKeyProvisioner(DeployKeyProvisioner):
                 body = json.loads(resp.read())
         except urllib.error.HTTPError as exc:
             _body = _read_error_body(exc)
+            if exc.code == 422:
+                raise DeployKeyCollisionError(
+                    f"deploy key collision for {owner_repo!r} "
+                    f"(title={title!r}): key title or content already registered — {_body}"
+                ) from exc
             raise RuntimeError(
                 f"failed to create deploy key for {owner_repo}: "
                 f"HTTP {exc.code} — {_body}"

bot_bottle/contrib/pi/agent_provider.py

@@ -21,6 +21,7 @@ from ...agent_provider import (
     AgentProvisionDir,
     AgentProvisionFile,
     AgentProvisionPlan,
+    provider_startup_args,
 )
 from ...egress import EgressRoute
 from ...log import die, info
@@ -165,7 +166,6 @@ _RUNTIME = AgentProviderRuntime(
     prompt_mode="append_system_prompt",
     bypass_args=(),
     resume_args=(),
-    remote_control_args=(),
 )
 
 
@@ -199,6 +199,7 @@ class PiAgentProvider(AgentProvider):
         models_payload, base_url, api_key_env, models, provider_name = (
             _pi_models_json(settings)
         )
+        extra_startup_args = provider_startup_args(provider_settings)
         models_file = state_dir / "pi-models.json"
         models_file.write_text(json.dumps(models_payload, indent=2) + "\n")
         models_file.chmod(0o600)
@@ -219,6 +220,7 @@ class PiAgentProvider(AgentProvider):
             startup_args=(
                 "--models",
                 ",".join(f"{provider_name}/{model}" for model in models),
+                *extra_startup_args,
             ),
             dirs=(AgentProvisionDir(f"{guest_home}/.pi/agent"),),
             files=(AgentProvisionFile(models_file, _models_path(guest_home)),),

bot_bottle/deploy_key_provisioner.py

@@ -11,6 +11,10 @@ from __future__ import annotations
 from abc import ABC, abstractmethod
 
 
+class DeployKeyCollisionError(RuntimeError):
+    """Raised when a deploy key title or public key already exists on the repo."""
+
+
 class DeployKeyProvisioner(ABC):
     """Manages a single deploy-key lifecycle on a remote forge."""
 

bot_bottle/dlp_detectors.py

@@ -78,16 +78,27 @@ TOKEN_PATTERNS: tuple[tuple[str, re.Pattern[str]], ...] = (
 )
 
 
-def scan_token_patterns(text: str, *, location: str = "body") -> ScanResult | None:
+def scan_token_patterns(
+    text: str,
+    *,
+    location: str = "body",
+    safe_tokens: typing.AbstractSet[str] | None = None,
+) -> ScanResult | None:
     normalized = _normalize_text(text)
     for name, pattern in TOKEN_PATTERNS:
-        m = pattern.search(normalized)
+        for m in pattern.finditer(normalized):
-        if m is not None:
+            value = m.group(0)
+            # A value the supervisor has approved (PRD 0062) is no longer a
+            # block — keep scanning so a second, un-approved token in the
+            # same request is still caught.
+            if safe_tokens is not None and value in safe_tokens:
+                continue
             return ScanResult(
                 severity="block",
                 reason=f"{name} found in {location}",
                 location=location,
-                context=_snippet(text, m.start(), m.end()),
+                context=_snippet(normalized, m.start(), m.end()),
+                matched=value,
             )
     return None
 
@@ -155,6 +166,7 @@ def scan_known_secrets(
     *,
     location: str = "body",
     env: typing.Mapping[str, str] | None = None,
+    safe_tokens: typing.AbstractSet[str] | None = None,
 ) -> ScanResult | None:
     if env is None:
         return None
@@ -164,11 +176,17 @@ def scan_known_secrets(
         for variant in _encoded_variants(value):
             pos = text.find(variant)
             if pos >= 0:
+                # The supervisor approves the exact encoded variant found
+                # (PRD 0062); a different encoding of the same secret is a
+                # fresh block.
+                if safe_tokens is not None and variant in safe_tokens:
+                    continue
                 return ScanResult(
                     severity="block",
                     reason=f"provisioned secret from {key} found in {location}",
                     location=location,
                     context=_snippet(text, pos, pos + len(variant)),
+                    matched=variant,
                 )
     return None
 
@@ -265,6 +283,14 @@ _CRLF_ENCODED_RE = re.compile(r"%0[dD]%0[aA]", re.ASCII)
 _CRLF_HEADER_INJECT_RE = re.compile(r"\r\n[A-Za-z][A-Za-z0-9\-]+\s*:", re.ASCII)
 
 
+def strip_crlf(text: str) -> str:
+    """Remove URL-encoded and literal CRLF injection sequences from a request
+    surface (PRD 0062 redact policy). Used to scrub the request line / headers
+    so the request can be forwarded instead of hard-blocked."""
+    text = _CRLF_ENCODED_RE.sub("", text)
+    return _CRLF_HEADER_INJECT_RE.sub(lambda m: m.group(0)[2:], text)
+
+
 def scan_crlf_injection(text: str) -> ScanResult | None:
     if _CRLF_ENCODED_RE.search(text):
         return ScanResult(
@@ -288,4 +314,5 @@ __all__ = [
     "scan_known_secrets",
     "scan_naive_injection",
     "scan_token_patterns",
+    "strip_crlf",
 ]

bot_bottle/egress.py

@@ -16,6 +16,7 @@ from pathlib import Path
 from typing import TYPE_CHECKING
 
 from .egress_addon_core import (
+    ON_MATCH_REDACT,
     HeaderMatch as CoreHeaderMatch,
     MatchEntry as CoreMatchEntry,
     PathMatch as CorePathMatch,
@@ -95,6 +96,7 @@ def egress_manifest_routes(
             git_fetch=r.GitFetch,
             outbound_detectors=r.OutboundDetectors,
             inbound_detectors=r.InboundDetectors,
+            outbound_on_match=r.OutboundOnMatch,
         ))
     return tuple(out)
 
@@ -105,12 +107,27 @@ def egress_routes_for_bottle(
 ) -> tuple[EgressRoute, ...]:
     manifest = egress_manifest_routes(bottle)
     provisioned_hosts = {pr.host.lower() for pr in provider_routes}
-    merged = list(provider_routes) + [
+    merged = list(_default_provider_on_match(provider_routes)) + [
         r for r in manifest if r.host.lower() not in provisioned_hosts
     ]
     return _assign_token_slots(merged)
 
 
+def _default_provider_on_match(
+    provider_routes: tuple[EgressRoute, ...],
+) -> tuple[EgressRoute, ...]:
+    """Provider routes (the agent talking to its own LLM API) default to the
+    `redact` on-match policy (PRD 0062): high-volume conversation payloads are
+    the worst source of token-shaped false positives, so a match is scrubbed
+    and forwarded rather than hard-blocked or queued for the operator. A
+    provider that sets `outbound_on_match` explicitly keeps its choice."""
+    return tuple(
+        r if r.outbound_on_match
+        else dataclasses.replace(r, outbound_on_match=ON_MATCH_REDACT)
+        for r in provider_routes
+    )
+
+
 def _assign_token_slots(
     routes: list[EgressRoute],
 ) -> tuple[EgressRoute, ...]:
@@ -177,7 +194,11 @@ def _route_to_yaml_fields(r: Route) -> dict[str, object]:
         fields["matches"] = matches_data
     if r.git_fetch:
         fields["git"] = {"fetch": True}
-    if r.outbound_detectors is not None or r.inbound_detectors is not None:
+    if (
+        r.outbound_detectors is not None
+        or r.inbound_detectors is not None
+        or r.outbound_on_match
+    ):
         dlp: dict[str, object] = {}
         if r.outbound_detectors is not None:
             dlp["outbound_detectors"] = (
@@ -189,6 +210,8 @@ def _route_to_yaml_fields(r: Route) -> dict[str, object]:
                 False if not r.inbound_detectors
                 else list(r.inbound_detectors)
             )
+        if r.outbound_on_match:
+            dlp["outbound_on_match"] = r.outbound_on_match
         fields["dlp"] = dlp
     return fields
 
@@ -260,6 +283,8 @@ def egress_render_routes(
                 elif isinstance(dv, list):
                     items_str = ", ".join(f'"{x}"' for x in dv)
                     lines.append(f"      {dk}: [{items_str}]")
+                elif isinstance(dv, str):
+                    lines.append(f'      {dk}: "{dv}"')
     return "\n".join(lines) + "\n"
 
 

bot_bottle/egress_addon.py

@@ -5,6 +5,7 @@ egress container."""
 
 from __future__ import annotations
 
+import asyncio
 import json
 import os
 import signal
@@ -16,9 +17,15 @@ from mitmproxy import http  # type: ignore[import-not-found]  # pylint: disable=
 from egress_addon_core import (  # type: ignore[import-not-found]  # pylint: disable=import-error
     LOG_BLOCKS,
     LOG_FULL,
+    DEFAULT_OUTBOUND_ON_MATCH,
+    ON_MATCH_BLOCK,
+    ON_MATCH_REDACT,
     Config,
+    Route,
+    ScanResult,
     build_inbound_scan_text,
     build_outbound_scan_text,
+    build_token_allow_payload,
     decide,
     decide_git_fetch,
     is_git_fetch_request,
@@ -32,23 +39,55 @@ from egress_addon_core import (  # type: ignore[import-not-found]  # pylint: dis
 )
 
 try:
-    from dlp_detectors import redact_tokens  # type: ignore[import-not-found]
+    from dlp_detectors import redact_tokens, strip_crlf  # type: ignore[import-not-found]
 except ImportError:  # pragma: no cover - host-side path
-    from bot_bottle.dlp_detectors import redact_tokens  # type: ignore[import-not-found]
+    from bot_bottle.dlp_detectors import (  # type: ignore[import-not-found]
+        redact_tokens,
+        strip_crlf,
+    )
+
+try:
+    import supervise as _sv  # type: ignore[import-not-found]
+except ImportError:  # pragma: no cover - host-side path
+    from bot_bottle import supervise as _sv  # type: ignore[import-not-found]
 
 
 DEFAULT_ROUTES_PATH = "/etc/egress/routes.yaml"
 
 INTROSPECT_HOST = "_egress.local"
 
+# Seconds the egress proxy holds a token-blocked request open waiting for the
+# operator's supervisor decision (PRD 0062), overridable via env.
+DEFAULT_TOKEN_ALLOW_TIMEOUT_SECONDS = 300.0
+# Filesystem poll cadence while awaiting the operator's response.
+TOKEN_ALLOW_POLL_INTERVAL_SECONDS = 0.5
+
+# Fixed operator guidance attached to every token-allow proposal.
+_TOKEN_ALLOW_JUSTIFICATION = (
+    "egress DLP blocked an outbound request carrying a detected token. "
+    "Approve only if this value is a false positive or a credential this "
+    "request legitimately needs; the value is then allowed for the life of "
+    "this bottle's egress proxy."
+)
+
 
 class EgressAddon:
     def __init__(self) -> None:
         self.routes_path = os.environ.get("EGRESS_ROUTES", DEFAULT_ROUTES_PATH)
         self.config: Config = Config(routes=())
+        # Tokens the operator has approved this session (PRD 0062). In-memory
+        # only — a restart re-prompts. Mutated only from the asyncio loop that
+        # runs the addon hooks, so no lock is needed.
+        self.safe_tokens: set[str] = set()
+        self._supervise_queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "").strip()
+        self._supervise_slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "").strip()
+        self._token_allow_timeout = _token_allow_timeout_from_env(os.environ)
         self._reload(initial=True)
         self._install_sighup()
 
+    def _supervise_available(self) -> bool:
+        return bool(self._supervise_queue_dir and self._supervise_slug)
+
     def _reload(self, *, initial: bool = False) -> None:
         try:
             text = Path(self.routes_path).read_text(encoding="utf-8")
@@ -145,7 +184,7 @@ class EgressAddon:
             + "\n"
         )
 
-    def request(self, flow: http.HTTPFlow) -> None:
+    async def request(self, flow: http.HTTPFlow) -> None:
         request_path, _, query = flow.request.path.partition("?")
 
         if flow.request.pretty_host == INTROSPECT_HOST:
@@ -157,21 +196,11 @@ class EgressAddon:
         # Hostname is included to catch DNS-tunnelling exfiltration attempts.
         route = match_route(self.config.routes, flow.request.pretty_host)
         if route is not None:
-            body = flow.request.get_text(strict=False) or ""
+            if not await self._handle_outbound_dlp(flow, route):
-            scan_text = build_outbound_scan_text(
-                flow.request.pretty_host,
-                request_path,
-                query,
-                outbound_scan_headers(route, dict(flow.request.headers)),
-                body,
-            )
-            dlp_result = scan_outbound(route, scan_text, os.environ)
-            if dlp_result is not None and dlp_result.severity == "block":
-                ctx = self._req_ctx(flow)
-                if dlp_result.context:
-                    ctx = {**ctx, "context": dlp_result.context}
-                self._block(flow, f"egress DLP: {dlp_result.reason}", ctx=ctx)
                 return
+            # The redact policy may have rewritten the request line; recompute
+            # the path/query the git checks below rely on.
+            request_path, _, query = flow.request.path.partition("?")
 
         if is_git_push_request(request_path, query):
             self._block(
@@ -221,6 +250,202 @@ class EgressAddon:
         if self.config.log >= LOG_FULL:
             self._log_request(flow)
 
+    def _block_dlp(self, flow: http.HTTPFlow, result: ScanResult) -> None:
+        ctx = self._req_ctx(flow)
+        if result.context:
+            ctx = {**ctx, "context": result.context}
+        self._block(flow, f"egress DLP: {result.reason}", ctx=ctx)
+
+    async def _handle_outbound_dlp(
+        self,
+        flow: http.HTTPFlow,
+        route: Route,
+    ) -> bool:
+        """Scan the outbound request and apply the route's on-match policy
+        (PRD 0062). Returns True if the request may be forwarded, False if a
+        403 response has been written to `flow`.
+
+        Loops so the supervise policy can re-scan after each approval — a
+        second, un-approved token in the same request is still caught."""
+        while True:
+            request_path, _, query = flow.request.path.partition("?")
+            body = flow.request.get_text(strict=False) or ""
+            headers = outbound_scan_headers(route, dict(flow.request.headers))
+            scan_text = build_outbound_scan_text(
+                flow.request.pretty_host, request_path, query, headers, body,
+            )
+            # CRLF is scanned only over the request line + headers, never the
+            # body (see scan_outbound) — a body is not an injection vector.
+            crlf_text = build_outbound_scan_text(
+                flow.request.pretty_host, request_path, query, headers, "",
+            )
+            result = scan_outbound(
+                route, scan_text, os.environ,
+                safe_tokens=self.safe_tokens, crlf_text=crlf_text,
+            )
+            if result is None or result.severity != "block":
+                return True
+
+            policy = route.outbound_on_match or DEFAULT_OUTBOUND_ON_MATCH
+
+            # redact scrubs every detection (tokens and structural CRLF) and
+            # forwards; it fails closed only if a match survives the scrub.
+            if policy == ON_MATCH_REDACT:
+                if self._redact_outbound(flow, route):
+                    if self.config.log >= LOG_BLOCKS:
+                        sys.stderr.write(json.dumps({
+                            "event": "egress_redacted",
+                            "reason": f"egress DLP: {result.reason}",
+                            **self._req_ctx(flow),
+                        }) + "\n")
+                    return True
+                self._block(
+                    flow,
+                    f"egress DLP: {result.reason}; redaction could not remove "
+                    "all matches (e.g. a match in the hostname)",
+                    ctx=self._req_ctx(flow),
+                )
+                return False
+
+            # Structural blocks (CRLF, no safelist-able value) cannot be
+            # supervised — there is nothing to approve and remember — so under
+            # block/supervise they are a hard 403.
+            if policy == ON_MATCH_BLOCK or not result.matched:
+                self._block_dlp(flow, result)
+                return False
+
+            # supervise (default): hold the request for operator approval.
+            # Fall back to a hard 403 when supervise isn't wired for the bottle.
+            if not self._supervise_available():
+                self._block_dlp(flow, result)
+                return False
+            approved = await self._supervise_token_block(flow, request_path, result)
+            if not approved:
+                return False  # _supervise_token_block wrote the 403 response
+            # loop: the approved value is now in safe_tokens; re-scan.
+
+    def _redact_outbound(self, flow: http.HTTPFlow, route: Route) -> bool:
+        """Scrub detected tokens (and CRLF injection sequences) from the mutable
+        request surfaces (body, headers, path/query) and re-scan. Returns True
+        if the request is now clean; False if a block-severity match remains on
+        a surface redaction cannot rewrite (the hostname) so the caller fails
+        closed."""
+        body = flow.request.get_text(strict=False)
+        if body:
+            redacted_body = redact_tokens(body, env=os.environ)
+            if redacted_body != body:
+                flow.request.text = redacted_body
+        for name, value in list(flow.request.headers.items()):
+            if name.lower() == "host":
+                continue  # routing-critical; never a legitimate token
+            redacted = strip_crlf(redact_tokens(value, env=os.environ))
+            if redacted != value:
+                flow.request.headers[name] = redacted
+        redacted_path = strip_crlf(redact_tokens(flow.request.path, env=os.environ))
+        if redacted_path != flow.request.path:
+            flow.request.path = redacted_path
+
+        request_path, _, query = flow.request.path.partition("?")
+        new_body = flow.request.get_text(strict=False) or ""
+        headers = outbound_scan_headers(route, dict(flow.request.headers))
+        scan_text = build_outbound_scan_text(
+            flow.request.pretty_host, request_path, query, headers, new_body,
+        )
+        crlf_text = build_outbound_scan_text(
+            flow.request.pretty_host, request_path, query, headers, "",
+        )
+        result = scan_outbound(route, scan_text, os.environ, crlf_text=crlf_text)
+        return result is None or result.severity != "block"
+
+    async def _supervise_token_block(
+        self,
+        flow: http.HTTPFlow,
+        request_path: str,
+        result: ScanResult,
+    ) -> bool:
+        """Route a token DLP block to the operator's supervisor queue and wait.
+
+        Returns True if the operator approved (the matched value is added to
+        `self.safe_tokens` and the caller re-scans); False if the request must
+        be blocked (a 403 response has been written to `flow`)."""
+        host = flow.request.pretty_host
+        payload = build_token_allow_payload(
+            redact_tokens(host, env=os.environ),
+            flow.request.method,
+            redact_tokens(request_path, env=os.environ),
+            result,
+        )
+        proposal = _sv.Proposal.new(
+            bottle_slug=self._supervise_slug,
+            tool=_sv.TOOL_EGRESS_TOKEN_ALLOW,
+            proposed_file=payload,
+            justification=_TOKEN_ALLOW_JUSTIFICATION,
+            current_file_hash=_sv.sha256_hex(payload),
+        )
+        queue_dir = Path(self._supervise_queue_dir)
+        try:
+            _sv.write_proposal(queue_dir, proposal)
+        except OSError as e:
+            sys.stderr.write(
+                f"egress: could not queue token-allow proposal: {e}; "
+                "blocking request\n"
+            )
+            self._block(flow, f"egress DLP: {result.reason}", ctx=self._req_ctx(flow))
+            return False
+
+        sys.stderr.write(json.dumps({
+            "event": "egress_token_supervise",
+            "reason": f"egress DLP: {result.reason}",
+            "proposal": proposal.id,
+            **self._req_ctx(flow),
+        }) + "\n")
+
+        response = await self._await_token_response(queue_dir, proposal.id)
+        _sv.archive_proposal(queue_dir, proposal.id)
+
+        if response is not None and response.status in (
+            _sv.STATUS_APPROVED, _sv.STATUS_MODIFIED,
+        ):
+            self.safe_tokens.add(result.matched)
+            if self.config.log >= LOG_BLOCKS:
+                sys.stderr.write(json.dumps({
+                    "event": "egress_token_allowed",
+                    "reason": f"egress DLP: {result.reason}",
+                    "proposal": proposal.id,
+                    **self._req_ctx(flow),
+                }) + "\n")
+            return True
+
+        if response is None:
+            reason = (
+                f"egress DLP: {result.reason}; supervisor approval timed out "
+                f"after {self._token_allow_timeout:g}s"
+            )
+        else:
+            reason = f"egress DLP: {result.reason}; supervisor rejected the request"
+        self._block(flow, reason, ctx=self._req_ctx(flow))
+        return False
+
+    async def _await_token_response(
+        self,
+        queue_dir: Path,
+        proposal_id: str,
+    ) -> "_sv.Response | None":
+        """Poll the queue dir for the operator's response without blocking the
+        proxy event loop. Returns the Response, or None on timeout."""
+        loop = asyncio.get_running_loop()
+        deadline = loop.time() + self._token_allow_timeout
+        while True:
+            try:
+                return _sv.read_response(queue_dir, proposal_id)
+            except (OSError, ValueError, KeyError):
+                # Not written yet, or a partial/malformed write — retry until
+                # the deadline, then fail closed.
+                pass
+            if loop.time() >= deadline:
+                return None
+            await asyncio.sleep(TOKEN_ALLOW_POLL_INTERVAL_SECONDS)
+
     def response(self, flow: http.HTTPFlow) -> None:
         """DLP inbound scan on response headers and body."""
         route = match_route(self.config.routes, flow.request.pretty_host)
@@ -272,7 +497,12 @@ class EgressAddon:
         message = flow.websocket.messages[-1]  # type: ignore[union-attr]
         content = message.content.decode("utf-8", errors="replace")
         if message.from_client:
-            result = scan_outbound(route, content, os.environ)
+            # A WebSocket data frame is not an HTTP request line, so CRLF is
+            # not an injection vector here — scan only for credential leakage.
+            result = scan_outbound(
+                route, content, os.environ,
+                safe_tokens=self.safe_tokens, crlf_text="",
+            )
             if result is not None and result.severity == "block":
                 sys.stderr.write(f"egress DLP: {result.reason}\n")
                 flow.kill()  # type: ignore[union-attr]
@@ -286,4 +516,23 @@ class EgressAddon:
                     sys.stderr.write(f"egress DLP warn: {result.reason}\n")
 
 
+def _token_allow_timeout_from_env(env: "os._Environ[str]") -> float:
+    """Read EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS; fall back to the default on an
+    unset or invalid value (a bad value should not wedge egress at boot)."""
+    raw = env.get("EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS", "").strip()
+    if not raw:
+        return DEFAULT_TOKEN_ALLOW_TIMEOUT_SECONDS
+    try:
+        value = float(raw)
+    except ValueError:
+        value = 0.0
+    if value <= 0:
+        sys.stderr.write(
+            "egress: invalid EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS="
+            f"{raw!r}; using default {DEFAULT_TOKEN_ALLOW_TIMEOUT_SECONDS:g}s\n"
+        )
+        return DEFAULT_TOKEN_ALLOW_TIMEOUT_SECONDS
+    return value
+
+
 addons = [EgressAddon()]

bot_bottle/egress_addon_core.py

@@ -37,6 +37,15 @@ VALID_METHODS = frozenset({
 OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets"})
 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:
@@ -69,6 +78,8 @@ class Route:
     git_fetch: bool = False
     outbound_detectors: tuple[str, ...] | None = None
     inbound_detectors: tuple[str, ...] | None = None
+    # "" means unset → DEFAULT_OUTBOUND_ON_MATCH. See OUTBOUND_ON_MATCH_VALUES.
+    outbound_on_match: str = ""
 
 
 LOG_OFF = 0    # no logging
@@ -95,6 +106,11 @@ class ScanResult:
     reason: str
     location: str = ""  # where the match was found, e.g. "body", "authorization header"
     context: str = ""   # surrounding text with the match replaced by REDACT
+    # Raw substring the detector matched. Used inside the sidecar to key the
+    # supervisor-approved "safe tokens" set (PRD 0062); never logged or written
+    # to a proposal file. Empty for structural detectors (CRLF) that carry no
+    # safelist-able value.
+    matched: str = ""
 
 
 # ---------------------------------------------------------------------------
@@ -218,12 +234,12 @@ def _parse_detectors(
     idx: int,
     host: str,
     raw_dict: dict[str, object],
-) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None]:
+) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
     """Parse the optional `dlp` block on a route, returning
-    (outbound_detectors, inbound_detectors)."""
+    (outbound_detectors, inbound_detectors, outbound_on_match)."""
     dlp_raw = raw_dict.get("dlp")
     if dlp_raw is None:
-        return None, None
+        return None, None, ""
     label = f"route[{idx}] ({host})"
     if not isinstance(dlp_raw, dict):
         raise ValueError(f"{label}: 'dlp' must be an object")
@@ -260,13 +276,24 @@ def _parse_detectors(
     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"):
+        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"are 'outbound_detectors', 'inbound_detectors', "
+                f"'outbound_on_match'"
             )
-    return outbound, inbound
+    return outbound, inbound, on_match
 
 
 def parse_routes(payload: object) -> tuple[Route, ...]:
@@ -337,7 +364,7 @@ def _parse_one(idx: int, raw: object) -> Route:
                 )
 
     # dlp detectors
-    outbound_detectors, inbound_detectors = _parse_detectors(
+    outbound_detectors, inbound_detectors, outbound_on_match = _parse_detectors(
         idx, host, raw_dict,
     )
 
@@ -356,6 +383,7 @@ def _parse_one(idx: int, raw: object) -> Route:
         git_fetch=git_fetch,
         outbound_detectors=outbound_detectors,
         inbound_detectors=inbound_detectors,
+        outbound_on_match=outbound_on_match,
     )
 
 
@@ -404,6 +432,8 @@ def route_to_yaml_dict(r: Route) -> dict[str, object]:
         dlp["outbound_detectors"] = list(r.outbound_detectors)
     if r.inbound_detectors is not None:
         dlp["inbound_detectors"] = list(r.inbound_detectors)
+    if r.outbound_on_match:
+        dlp["outbound_on_match"] = r.outbound_on_match
     if dlp:
         d["dlp"] = dlp
     return d
@@ -690,6 +720,9 @@ def scan_outbound(
     route: Route,
     body: str | bytes,
     environ: typing.Mapping[str, str],
+    *,
+    safe_tokens: typing.AbstractSet[str] | None = None,
+    crlf_text: str | None = None,
 ) -> ScanResult | None:
     # Lazy import to avoid circular deps and keep dlp_detectors optional
     # at import time (the sidecar copies it flat alongside this file).
@@ -708,25 +741,53 @@ def scan_outbound(
 
     text = body if isinstance(body, str) else body.decode("utf-8", errors="replace")
 
-    # CRLF injection is never legitimate — runs unconditionally, not gated
+    # CRLF injection is only an attack in the request line + headers, never the
-    # by outbound_detectors config.
+    # body: an HTTP body is delimited by Content-Length, so CRLF bytes there
-    result = scan_crlf_injection(text)
+    # cannot split the request. Scanning the body produces false positives on
+    # legitimate form-encoded / multi-line content. Callers pass the
+    # body-excluded surfaces as `crlf_text`; `None` falls back to the full text
+    # for backward-compatible callers (host-side tests, websocket frames).
+    crlf_target = text if crlf_text is None else crlf_text
+    result = scan_crlf_injection(crlf_target)
     if result is not None:
         return result
 
     if _detector_enabled(route.outbound_detectors, "token_patterns"):
-        result = scan_token_patterns(text, location="body")
+        result = scan_token_patterns(text, location="body", safe_tokens=safe_tokens)
         if result is not None:
             return result
 
     if _detector_enabled(route.outbound_detectors, "known_secrets"):
-        result = scan_known_secrets(text, location="body", env=environ)
+        result = scan_known_secrets(
+            text, location="body", env=environ, safe_tokens=safe_tokens,
+        )
         if result is not None:
             return result
 
     return None
 
 
+def build_token_allow_payload(
+    host: str,
+    method: str,
+    path: str,
+    result: ScanResult,
+) -> str:
+    """Render the human-readable supervisor proposal body for an outbound
+    token block (PRD 0062). Carries the host/method/path, the detector
+    reason, and the redacted context snippet — never the raw token value."""
+    lines = [
+        "egress blocked an outbound request carrying a detected token",
+        f"host: {host}",
+        f"method: {method}",
+        f"path: {path}",
+        f"detector: {result.reason}",
+    ]
+    if result.context:
+        lines.append(f"context: {result.context}")
+    return "\n".join(lines) + "\n"
+
+
 def scan_inbound(
     route: Route,
     body: str | bytes,
@@ -751,6 +812,11 @@ __all__ = [
     "route_to_yaml_dict",
     "LOG_FULL",
     "LOG_OFF",
+    "ON_MATCH_BLOCK",
+    "ON_MATCH_REDACT",
+    "ON_MATCH_SUPERVISE",
+    "OUTBOUND_ON_MATCH_VALUES",
+    "DEFAULT_OUTBOUND_ON_MATCH",
     "Config",
     "Decision",
     "HeaderMatch",
@@ -760,6 +826,7 @@ __all__ = [
     "ScanResult",
     "build_inbound_scan_text",
     "build_outbound_scan_text",
+    "build_token_allow_payload",
     "decide",
     "decide_git_fetch",
     "evaluate_matches",

bot_bottle/git_gate.py

@@ -247,6 +247,164 @@ cat > "$refs_file"
 
 zero=0000000000000000000000000000000000000000
 
+supervise_gitleaks_allow() {
+  log_opts=$1
+  ref=$2
+  report_file=$(mktemp)
+  if ! gitleaks git \
+      --log-opts="$log_opts" \
+      --no-banner \
+      --redact \
+      --ignore-gitleaks-allow \
+      --report-format=json \
+      --report-path="$report_file" \
+      --exit-code 0 \
+      1>&2; then
+    rm -f "$report_file"
+    echo "git-gate: gitleaks inline-suppression scan failed for $ref" >&2
+    return 1
+  fi
+
+  proposal_id=$(
+    GITLEAKS_ALLOW_REF="$ref" python3 - "$report_file" <<'PY'
+import datetime
+import hashlib
+import json
+import os
+import sys
+import uuid
+from pathlib import Path
+
+report_path = Path(sys.argv[1])
+queue_dir = os.environ.get("SUPERVISE_QUEUE_DIR", "")
+slug = os.environ.get("SUPERVISE_BOTTLE_SLUG", "")
+if not queue_dir or not slug:
+    sys.exit(2)
+
+try:
+    raw = json.loads(report_path.read_text() or "[]")
+except json.JSONDecodeError:
+    sys.exit(3)
+if not isinstance(raw, list):
+    sys.exit(3)
+if not raw:
+    sys.exit(0)
+
+ref = os.environ.get("GITLEAKS_ALLOW_REF", "")
+lines = [
+    "gitleaks inline suppression requires supervisor approval",
+    f"ref: {ref}",
+    "",
+]
+for i, finding in enumerate(raw, 1):
+    if not isinstance(finding, dict):
+        continue
+    file_path = finding.get("File", "")
+    line_no = finding.get("StartLine", finding.get("Line", ""))
+    rule_id = finding.get("RuleID", "")
+    commit = finding.get("Commit", "")
+    line = finding.get("Line", "")
+    lines.extend([
+        f"finding {i}:",
+        f"  file: {file_path}",
+        f"  line: {line_no}",
+        f"  rule: {rule_id}",
+        f"  commit: {commit}",
+        f"  code: {line}",
+        "",
+    ])
+
+payload = "\n".join(lines).rstrip() + "\n"
+proposal_id = str(uuid.uuid4())
+proposal = {
+    "id": proposal_id,
+    "bottle_slug": slug,
+    "tool": "gitleaks-allow",
+    "proposed_file": payload,
+    "justification": (
+        "git-gate found gitleaks findings hidden by # gitleaks:allow; "
+        "approve only for dummy test fixtures or confirmed false positives"
+    ),
+    "arrival_timestamp": datetime.datetime.now(
+        datetime.timezone.utc
+    ).isoformat(),
+    "current_file_hash": hashlib.sha256(payload.encode("utf-8")).hexdigest(),
+}
+queue = Path(queue_dir)
+queue.mkdir(parents=True, exist_ok=True)
+path = queue / f"{proposal_id}.proposal.json"
+tmp = path.with_suffix(path.suffix + ".tmp")
+with tmp.open("w", encoding="utf-8") as f:
+    json.dump(proposal, f, indent=2)
+    f.write("\n")
+os.chmod(tmp, 0o600)
+os.replace(tmp, path)
+print(proposal_id)
+PY
+  )
+  rc=$?
+  rm -f "$report_file"
+  if [ "$rc" -eq 0 ] && [ -z "$proposal_id" ]; then
+    return 0
+  fi
+  if [ "$rc" -ne 0 ]; then
+    echo "git-gate: cannot route # gitleaks:allow finding to supervisor; refusing push" >&2
+    return 1
+  fi
+
+  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]*)
+      echo "git-gate: invalid SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS=$timeout" >&2
+      return 1
+      ;;
+  esac
+  echo "git-gate: queued # gitleaks:allow supervisor approval $proposal_id" >&2
+  echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
+  waited=0
+  while [ "$waited" -lt "$timeout" ]; do
+    if [ -f "$response_file" ]; then
+      status=$(python3 - "$response_file" <<'PY'
+import json
+import sys
+try:
+    with open(sys.argv[1], encoding="utf-8") as f:
+        raw = json.load(f)
+except (OSError, json.JSONDecodeError):
+    sys.exit(1)
+status = raw.get("status")
+if not isinstance(status, str):
+    sys.exit(1)
+print(status)
+PY
+      ) || status=""
+      case "$status" in
+        approved|modified)
+          mkdir -p "$queue_dir/processed"
+          mv -f "$queue_dir/${proposal_id}.proposal.json" "$queue_dir/processed/" 2>/dev/null || true
+          mv -f "$queue_dir/${proposal_id}.response.json" "$queue_dir/processed/" 2>/dev/null || true
+          echo "git-gate: supervisor approved # gitleaks:allow for $ref" >&2
+          return 0
+          ;;
+        rejected)
+          echo "git-gate: supervisor rejected # gitleaks:allow for $ref" >&2
+          return 1
+          ;;
+        *)
+          echo "git-gate: invalid supervisor response for # gitleaks:allow" >&2
+          return 1
+          ;;
+      esac
+    fi
+    sleep 1
+    waited=$((waited + 1))
+  done
+  echo "git-gate: supervisor approval timed out for # gitleaks:allow; refusing push" >&2
+  return 1
+}
+
 # Phase 1: gitleaks scan each ref's incoming commits.
 while IFS=' ' read -r old new ref; do
   [ -z "$ref" ] && continue
@@ -268,6 +426,9 @@ while IFS=' ' read -r old new ref; do
     echo "git-gate: gitleaks rejected push to $ref" >&2
     exit 1
   fi
+  if ! supervise_gitleaks_allow "$log_opts" "$ref"; then
+    exit 1
+  fi
 done < "$refs_file"
 
 # Phase 2: forward each ref to the upstream (`origin`, configured

bot_bottle/log.py

@@ -1,21 +1,107 @@
-"""Tiny logging wrappers. All output goes to stderr."""
+"""Tiny logging wrappers. All output goes to stderr.
+
+Two capabilities layer onto the bare wrappers (issue #252):
+
+  - **Levels.** `debug` / `info` / `warn` / `error` carry an ordered
+    severity. Output is gated by `BOT_BOTTLE_LOG_LEVEL` (debug | info |
+    warn | error; default `info`). A message emits when its severity is
+    at or above the threshold, so `debug` is silent by default and
+    `error` always surfaces (nothing sits above it) — which keeps the
+    fatal `die` path visible regardless of the configured level.
+
+  - **Context.** Every wrapper takes an optional `context` mapping that
+    renders as a parseable ` [k=v ...]` suffix (keys sorted; values with
+    whitespace/quotes are quoted), so failures can be filtered and
+    correlated instead of being flat strings.
+
+With no `context` and the default level, output is byte-identical to the
+original `bot-bottle: <msg>` / `bot-bottle: warning: <msg>` /
+`bot-bottle: error: <msg>` lines — the 100+ existing call sites are
+unaffected.
+"""
 
 from __future__ import annotations
 
+import os
 import sys
-from typing import NoReturn
+from typing import Mapping, NoReturn
+
+# Ordered severities. Gaps left between values so intermediate levels
+# can be added later without renumbering.
+DEBUG = 10
+INFO = 20
+WARN = 30
+ERROR = 40
+
+_LEVEL_NAMES: dict[str, int] = {
+    "debug": DEBUG,
+    "info": INFO,
+    "warn": WARN,
+    "warning": WARN,
+    "error": ERROR,
+}
+
+# Default threshold when BOT_BOTTLE_LOG_LEVEL is unset or unrecognised.
+_DEFAULT_THRESHOLD = INFO
+
+_LOG_LEVEL_ENV = "BOT_BOTTLE_LOG_LEVEL"
 
 
-def info(msg: str) -> None:
+def _threshold() -> int:
-    print(f"bot-bottle: {msg}", file=sys.stderr)
+    """Resolve the active level threshold from the environment.
+
+    Read per-call (not cached) so the level can be changed at runtime
+    and so tests can patch `os.environ` without a reload. Unknown values
+    fall back to the default rather than raising — logging must never be
+    the thing that crashes the process."""
+    raw = os.environ.get(_LOG_LEVEL_ENV, "")
+    return _LEVEL_NAMES.get(raw.strip().lower(), _DEFAULT_THRESHOLD)
 
 
-def warn(msg: str) -> None:
+def _format_context(context: Mapping[str, object] | None) -> str:
-    print(f"bot-bottle: warning: {msg}", file=sys.stderr)
+    """Render a context mapping as a ` [k=v k2=v2]` suffix.
+
+    Keys are sorted for stable, diffable output. Values that are empty or
+    contain whitespace or a quote are wrapped in double quotes (with inner
+    quotes escaped) so each `k=v` pair stays parseable. Empty/None context
+    renders as the empty string."""
+    if not context:
+        return ""
+    parts: list[str] = []
+    for key in sorted(context):
+        value = str(context[key])
+        if value == "" or any(ch.isspace() for ch in value) or '"' in value:
+            value = '"' + value.replace('"', '\\"') + '"'
+        parts.append(f"{key}={value}")
+    return " [" + " ".join(parts) + "]"
 
 
-def error(msg: str) -> None:
+def _emit(
-    print(f"bot-bottle: error: {msg}", file=sys.stderr)
+    level: int,
+    label: str,
+    msg: str,
+    context: Mapping[str, object] | None,
+) -> None:
+    if level < _threshold():
+        return
+    prefix = f"{label}: " if label else ""
+    sys.stderr.write(f"bot-bottle: {prefix}{msg}{_format_context(context)}\n")
+
+
+def debug(msg: str, *, context: Mapping[str, object] | None = None) -> None:
+    _emit(DEBUG, "debug", msg, context)
+
+
+def info(msg: str, *, context: Mapping[str, object] | None = None) -> None:
+    _emit(INFO, "", msg, context)
+
+
+def warn(msg: str, *, context: Mapping[str, object] | None = None) -> None:
+    _emit(WARN, "warning", msg, context)
+
+
+def error(msg: str, *, context: Mapping[str, object] | None = None) -> None:
+    _emit(ERROR, "error", msg, context)
 
 
 class Die(SystemExit):
@@ -31,6 +117,6 @@ class Die(SystemExit):
         self.message = message
 
 
-def die(msg: str) -> NoReturn:
+def die(msg: str, *, context: Mapping[str, object] | None = None) -> NoReturn:
-    error(msg)
+    error(msg, context=context)
     raise Die(1, msg)

bot_bottle/manifest.py

@@ -19,7 +19,7 @@ Bottle schema (frontmatter):
     repos:      { <name>: <git-gate-entry>, ... }  # optional
   egress: { routes: [ <egress-route>, ... ] }
     # route keys: host, matches, auth, role, dlp
-  supervise:    <bool>                          # optional
+  supervise:    <bool>                          # optional (default true)
 
 Agent schema (frontmatter):
   bottle:        <bottle-name>          # required
@@ -111,13 +111,13 @@ class ManifestBottle:
     # identity without any git-gate.repos upstreams, and vice versa.
     git_user: ManifestGitUser = field(default_factory=ManifestGitUser)
     egress: ManifestEgressConfig = field(default_factory=ManifestEgressConfig)
-    # Opt-in per-bottle stuck-recovery sidecar (PRD 0013). When true,
+    # Per-bottle stuck-recovery sidecar (PRD 0013). When true (the
-    # the launch step brings up a supervise sidecar that exposes MCP
+    # default, issue #249), the launch step brings up a supervise
-    # tools to the agent (egress-block, capability-block) plus mounts
+    # sidecar that exposes MCP tools to the agent (egress-block,
-    # the current-config dir read-only into the agent at
+    # capability-block) plus mounts the current-config dir read-only
-    # /etc/bot-bottle/current-config. False (the default) skips the
+    # into the agent at /etc/bot-bottle/current-config. Set
-    # sidecar and mount.
+    # `supervise: false` to skip the sidecar and mount.
-    supervise: bool = False
+    supervise: bool = True
 
     @classmethod
     def from_dict(cls, name: str, raw: object) -> "ManifestBottle":
@@ -190,7 +190,7 @@ class ManifestBottle:
             else ManifestEgressConfig()
         )
 
-        supervise_raw = d.get("supervise", False)
+        supervise_raw = d.get("supervise", True)
         if not isinstance(supervise_raw, bool):
             raise ManifestError(
                 f"bottle '{name}' supervise must be a boolean "

bot_bottle/manifest_agent.py

@@ -199,13 +199,10 @@ def _parse_provider_settings(
 ) -> dict[str, object]:
     if raw is None:
         return {}
-    if template != "pi":
-        raise ManifestError(
-            f"bottle '{bottle_name}' agent_provider.settings is only "
-            "supported for template 'pi'"
-        )
     settings = as_json_object(raw, f"bottle '{bottle_name}' agent_provider.settings")
-    allowed = {
+
+    common_allowed = {"startup_args"}
+    pi_allowed = {
         "provider",
         "base_url",
         "api",
@@ -218,12 +215,37 @@ def _parse_provider_settings(
         "supports_developer_role",
         "supports_reasoning_effort",
     }
+    if template == "pi":
+        allowed = common_allowed | pi_allowed
+    elif template in ("claude", "codex"):
+        allowed = common_allowed
+    elif template not in PROVIDER_TEMPLATES:
+        return dict(settings)
+    else:
+        allowed = common_allowed
+
     for key in settings:
         if key not in allowed:
             raise ManifestError(
                 f"bottle '{bottle_name}' agent_provider.settings has unknown "
                 f"key {key!r}; allowed: {', '.join(sorted(allowed))}"
             )
+    startup_args = settings.get("startup_args")
+    if startup_args is not None:
+        if not isinstance(startup_args, list):
+            raise ManifestError(
+                f"bottle '{bottle_name}' agent_provider.settings.startup_args "
+                f"must be an array of strings"
+            )
+        for i, arg in enumerate(startup_args):
+            if not isinstance(arg, str) or not arg:
+                raise ManifestError(
+                    f"bottle '{bottle_name}' agent_provider.settings."
+                    f"startup_args[{i}] must be a non-empty string"
+                )
+    if template != "pi":
+        return dict(settings)
+
     for key in ("provider", "base_url", "api", "api_key", "api_key_env"):
         value = settings.get(key)
         if value is not None and (not isinstance(value, str) or not value):

bot_bottle/manifest_egress.py

@@ -21,6 +21,9 @@ VALID_METHODS = frozenset({
 OUTBOUND_DETECTOR_NAMES = frozenset({"token_patterns", "known_secrets"})
 INBOUND_DETECTOR_NAMES = frozenset({"naive_injection_detection"})
 
+# What the proxy does on an outbound token match (PRD 0062).
+OUTBOUND_ON_MATCH_VALUES = ("block", "redact", "supervise")
+
 
 def validate_egress_routes(
     bottle_name: str,
@@ -67,6 +70,7 @@ class ManifestEgressRoute:
     GitFetch: bool = False
     OutboundDetectors: tuple[str, ...] | None = None
     InboundDetectors: tuple[str, ...] | None = None
+    OutboundOnMatch: str = ""
 
     @classmethod
     def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "ManifestEgressRoute":
@@ -161,8 +165,9 @@ class ManifestEgressRoute:
         # --- dlp ---
         outbound_detectors: tuple[str, ...] | None = None
         inbound_detectors: tuple[str, ...] | None = None
+        outbound_on_match = ""
         if "dlp" in d:
-            outbound_detectors, inbound_detectors = _parse_dlp_block(
+            outbound_detectors, inbound_detectors, outbound_on_match = _parse_dlp_block(
                 label, d.get("dlp"),
             )
 
@@ -201,6 +206,7 @@ class ManifestEgressRoute:
             GitFetch=git_fetch,
             OutboundDetectors=outbound_detectors,
             InboundDetectors=inbound_detectors,
+            OutboundOnMatch=outbound_on_match,
         )
 
 
@@ -323,7 +329,7 @@ def _parse_header_match(
 def _parse_dlp_block(
     route_label: str,
     raw: object,
-) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None]:
+) -> tuple[tuple[str, ...] | None, tuple[str, ...] | None, str]:
     label = f"{route_label} dlp"
     d = as_json_object(raw, label)
 
@@ -358,13 +364,24 @@ def _parse_dlp_block(
     outbound = _parse_field("outbound_detectors", OUTBOUND_DETECTOR_NAMES)
     inbound = _parse_field("inbound_detectors", INBOUND_DETECTOR_NAMES)
 
+    on_match = ""
+    on_match_raw = d.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 ManifestError(
+                f"{label} 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 d:
-        if k not in ("outbound_detectors", "inbound_detectors"):
+        if k not in ("outbound_detectors", "inbound_detectors", "outbound_on_match"):
             raise ManifestError(
                 f"{label} has unknown key {k!r}; accepted keys are "
-                f"'outbound_detectors', 'inbound_detectors'"
+                f"'outbound_detectors', 'inbound_detectors', "
+                f"'outbound_on_match'"
             )
-    return outbound, inbound
+    return outbound, inbound, on_match
 
 
 LOG_LEVELS = frozenset({0, 1, 2})

bot_bottle/supervise.py

@@ -50,12 +50,18 @@ SUPERVISE_PORT = 9100
 
 TOOL_CAPABILITY_BLOCK = "capability-block"
 TOOL_EGRESS_BLOCK = "egress-block"
-TOOL_ALLOW = "allow"
+TOOL_EGRESS_ALLOW = "egress-allow"
+TOOL_GITLEAKS_ALLOW = "gitleaks-allow"
+# Written directly by the egress addon (not an agent-facing MCP tool) when an
+# outbound DLP token block is routed to the operator for override (PRD 0062).
+TOOL_EGRESS_TOKEN_ALLOW = "egress-token-allow"
 TOOL_LIST_EGRESS_ROUTES = "list-egress-routes"
 TOOLS: tuple[str, ...] = (
-    TOOL_ALLOW,
+    TOOL_EGRESS_ALLOW,
     TOOL_CAPABILITY_BLOCK,
     TOOL_EGRESS_BLOCK,
+    TOOL_GITLEAKS_ALLOW,
+    TOOL_EGRESS_TOKEN_ALLOW,
     TOOL_LIST_EGRESS_ROUTES,
 )
 
@@ -74,7 +80,7 @@ EGRESS_INTROSPECT_URL = "http://_egress.local/allowlist"
 # here — those changes are captured by git history + the rebuild record
 # laid down in PRD 0016.
 COMPONENT_FOR_TOOL: dict[str, str] = {
-    TOOL_ALLOW: "egress",
+    TOOL_EGRESS_ALLOW: "egress",
     TOOL_EGRESS_BLOCK: "egress",
 }
 
@@ -553,6 +559,10 @@ __all__ = [
     "EGRESS_FORWARD_PROXY",
     "EGRESS_INTROSPECT_URL",
     "TOOL_CAPABILITY_BLOCK",
+    "TOOL_EGRESS_ALLOW",
+    "TOOL_EGRESS_BLOCK",
+    "TOOL_GITLEAKS_ALLOW",
+    "TOOL_EGRESS_TOKEN_ALLOW",
     "TOOL_LIST_EGRESS_ROUTES",
     "archive_proposal",
     "audit_dir",

bot_bottle/supervise_server.py

@@ -148,7 +148,7 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
             "allowlist. Returns JSON with one entry per allowed host, "
             "each carrying its matches rules (if any) and whether "
             "the proxy injects Authorization for the route. Use this "
-            "before composing an `allow` or `egress-block` proposal so "
+            "before composing an `egress-allow` or `egress-block` proposal so "
             "the new routes file extends the live one rather than "
             "replacing it."
         ),
@@ -159,7 +159,7 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
         },
     },
     {
-        "name": _sv.TOOL_ALLOW,
+        "name": _sv.TOOL_EGRESS_ALLOW,
         "description": (
             "Request operator approval to change the bottle's egress "
             "allowlist. Pass the full proposed routes.yaml content, not "
@@ -187,6 +187,7 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
                         "  dlp:  (optional DLP scanner overrides)\n"
                         "    outbound_detectors: [token_patterns, known_secrets]\n"
                         "    inbound_detectors: [naive_injection_detection]\n"
+                        "    outbound_on_match: block|redact|supervise  (default supervise)\n"
                         "Omit any key that should use its default. "
                         "`list-egress-routes` returns routes in this same format."
                     ),
@@ -228,6 +229,7 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
                         "  dlp:  (optional DLP scanner overrides)\n"
                         "    outbound_detectors: [token_patterns, known_secrets]\n"
                         "    inbound_detectors: [naive_injection_detection]\n"
+                        "    outbound_on_match: block|redact|supervise  (default supervise)\n"
                         "Omit any key that should use its default. "
                         "`list-egress-routes` returns routes in this same format."
                     ),
@@ -274,7 +276,7 @@ TOOL_DEFINITIONS: list[dict[str, object]] = [
 # Map each proposal tool to the input field that carries the agent's
 # payload (stored in Proposal.proposed_file).
 PROPOSED_FILE_FIELD: dict[str, str] = {
-    _sv.TOOL_ALLOW: "routes_yaml",
+    _sv.TOOL_EGRESS_ALLOW: "routes_yaml",
     _sv.TOOL_CAPABILITY_BLOCK: "dockerfile",
     _sv.TOOL_EGRESS_BLOCK: "routes_yaml",
 }
@@ -293,7 +295,7 @@ def validate_proposed_file(tool: str, content: str) -> None:
         # Dockerfiles are too varied to validate syntactically beyond
         # non-empty. The operator reads the diff in the TUI.
         pass
-    elif tool in (_sv.TOOL_ALLOW, _sv.TOOL_EGRESS_BLOCK):
+    elif tool in (_sv.TOOL_EGRESS_ALLOW, _sv.TOOL_EGRESS_BLOCK):
         try:
             load_routes(content)
         except ValueError as e:

docs/prds/0060-commit-bottle-state.md

@@ -1,4 +1,4 @@
-# PRD prd-new: Commit bottle state to an image
+# PRD 0060: Commit bottle state to an image
 
 - **Status:** Active
 - **Author:** Claude

docs/prds/0061-gitleaks-inline-suppression-supervision.md

@@ -0,0 +1,101 @@
+# PRD 0061: Gitleaks inline suppression supervision
+
+- **Status:** Active
+- **Author:** codex
+- **Created:** 2026-06-10
+- **Issue:** #208
+
+## Summary
+
+Route every `# gitleaks:allow` inline suppression through the existing
+supervisor approval queue before git-gate forwards a push upstream. The
+annotation remains available for test fixtures and confirmed false positives,
+but it no longer lets an agent silently bypass the gate's secret scan.
+
+## Problem
+
+Gitleaks supports inline suppression comments. With the default behaviour,
+a pushed line that contains a secret-like value and `# gitleaks:allow` is
+ignored by gitleaks and accepted by git-gate. That is convenient for dummy
+credentials in tests, but it also gives an agent a one-line bypass for a real
+secret. Once the commit lands, the diff alone cannot prove whether the
+suppression was legitimate.
+
+## Goals / Success Criteria
+
+1. Git-gate continues to run the normal gitleaks scan for every incoming ref.
+2. After the normal scan passes, git-gate runs a second scan with
+   `--ignore-gitleaks-allow` and a JSON report so suppressed findings become
+   visible.
+3. If that second scan reports no suppressed findings, the push proceeds
+   unchanged.
+4. If it reports suppressed findings, git-gate creates a `gitleaks-allow`
+   supervisor proposal containing the ref, file path, line number, rule,
+   commit, and flagged line for each finding.
+5. The push proceeds only when the supervisor explicitly approves the
+   proposal; rejection, malformed responses, missing supervisor configuration,
+   and timeout all refuse the push.
+6. The supervisor TUI requires a reason when approving a `gitleaks-allow`
+   proposal, so the audit trail records whether the approval was for a test
+   fixture or a false positive.
+
+## Non-goals
+
+- Replacing gitleaks or changing the main secret-detection rule set.
+- Removing support for `# gitleaks:allow`.
+- Automatically classifying fixture files or false positives.
+- Adding new supervisor transport or authentication mechanisms.
+
+## Design
+
+### Git-gate flow
+
+`git_gate_render_hook()` emits a `supervise_gitleaks_allow` shell helper.
+For each incoming ref, git-gate first runs the existing gitleaks command. If
+that scan passes, it runs:
+
+```sh
+gitleaks git \
+  --log-opts="$log_opts" \
+  --no-banner \
+  --redact \
+  --ignore-gitleaks-allow \
+  --report-format=json \
+  --report-path="$report_file" \
+  --exit-code 0
+```
+
+The second pass keeps the push path non-interactive while producing a report
+of findings that would otherwise have been hidden by inline suppression.
+
+### Supervisor proposal
+
+When the JSON report contains findings, an embedded Python helper writes a
+proposal into `SUPERVISE_QUEUE_DIR` using the existing proposal schema. The
+proposal uses:
+
+- `tool: "gitleaks-allow"`
+- a text payload with the ref and each finding's file, line, rule, commit,
+  and redacted code line
+- a justification that tells the operator to approve only dummy test fixtures
+  or confirmed false positives
+
+Git-gate then waits for `<proposal-id>.response.json` for
+`SUPERVISE_GITLEAKS_ALLOW_TIMEOUT_SECONDS`, defaulting to 300 seconds.
+`approved` and `modified` responses allow the push; `rejected`, invalid
+responses, invalid timeout configuration, or timeout refuse it.
+
+### Supervisor UI
+
+`TOOL_GITLEAKS_ALLOW` is added to the supervisor tool registry. The curses
+supervisor renders the proposal as text and allows approval or rejection.
+Modification is unavailable for this proposal type because there is no file
+patch to apply. Approval from the TUI prompts for a non-empty reason and
+writes that reason to the response/audit path.
+
+### Tests
+
+Unit tests assert that the rendered git-gate hook includes the second gitleaks
+pass, supervisor queue fields, and fail-closed messages. Supervisor tests cover
+the new tool constant, proposal archiving, and the required TUI approval
+reason.

docs/prds/0062-egress-supervisor-token-override.md

@@ -0,0 +1,210 @@
+# PRD 0062: Supervisor override for egress token blocks
+
+- **Status:** Active
+- **Author:** claude
+- **Created:** 2026-06-24
+- **Issue:** #261
+
+## Summary
+
+Give each egress route a policy for what happens when an outbound DLP detector
+matches a token, via `dlp.outbound_on_match: block | redact | supervise`
+(default `supervise`):
+
+- **`supervise`** (default) — route the block through the existing supervisor
+  approval queue instead of returning `403` immediately. The proxy holds the
+  request open until the operator approves or rejects it. On approval the
+  matched token is added to an in-memory "safe tokens" set so the request — and
+  any later request carrying the same token — flows through without
+  re-prompting.
+- **`redact`** — scrub the matched value(s) from the request and forward it,
+  no operator in the loop. For routes where a token-shaped value is noise the
+  upstream doesn't need (telemetry/log sinks). Fails closed if a match lands on
+  a surface redaction can't rewrite (the hostname).
+- **`block`** — the original hard `403`; never overridable. For routes where a
+  detected token must always stop.
+
+The motivating goal is reducing friction from false positives without weakening
+the default-deny posture: supervise keeps a human in the loop, redact is an
+explicit per-route opt-in, and block stays available for sensitive routes.
+
+## Problem
+
+The outbound DLP detectors (`token_patterns`, `known_secrets`) are
+deliberately aggressive: any string that looks like a credential is blocked
+before it leaves the bottle. That is the right default, but it produces false
+positives — a token-shaped value that is not actually a secret, or a credential
+the agent legitimately needs to send to a declared host. Today the only
+recovery is for the operator to notice the `egress DLP` 403 in the logs and
+hand-edit the route's `dlp.outbound_detectors`, which disables the detector for
+the whole route rather than allowing the one value.
+
+The operator has no in-the-loop signal that a token block happened and no
+fine-grained way to say "this specific value is fine."
+
+## Goals / Success Criteria
+
+1. An outbound DLP **token** block (a `ScanResult` carrying a matched secret
+   value) creates a supervisor proposal instead of an immediate `403`.
+2. The egress proxy holds the blocked request open, polling for the operator's
+   response up to a bounded timeout.
+3. The proposal shows the operator the host, method, path, the detector reason,
+   and a **redacted** context snippet — never the raw token value.
+4. On `approved`/`modified`, the matched token value is added to an in-memory
+   safe-tokens set and the request proceeds normally; later requests carrying
+   the same value skip the block.
+5. On `rejected`, timeout, malformed response, or missing supervisor wiring,
+   the request fails closed with the same `403` as today.
+6. Structural blocks that carry no token value (CRLF injection) and the
+   route-not-allowlisted / git blocks are unchanged — they stay hard `403`s and
+   keep their existing agent-driven `allow` / `egress-block` MCP path.
+7. The proxy event loop is not stalled while waiting: the wait is asynchronous,
+   so other flows keep being served.
+
+## Non-goals
+
+- Persisting the safe-tokens set across egress restarts. It lives in process
+  memory only; a restart re-prompts. (The issue explicitly defers persistence.)
+- Supervising inbound (prompt-injection) blocks or WebSocket frame blocks.
+  WebSocket frames still honour the safe-tokens set for already-approved values
+  but cannot wait for approval (there is no response surface after upgrade).
+- Generalising an approved secret across encodings. The safe-tokens set matches
+  the exact value the detector found.
+- Replacing the per-route `dlp.outbound_detectors` override. That remains the
+  way to turn a detector off wholesale.
+- Making `redact` the default. Silent redaction of a true false positive
+  corrupts legitimate data, so it is opt-in per route; `supervise` (human in
+  the loop) stays the default.
+
+## Scope
+
+### In scope
+
+The minimum cut that ships, in build order:
+
+1. **Core** — `ScanResult.matched`; thread `safe_tokens` through
+   `scan_outbound` / the token detectors; `build_token_allow_payload`.
+2. **Supervise + TUI** — `TOOL_EGRESS_TOKEN_ALLOW`; TUI suffix, modify guard,
+   required approval reason.
+3. **Addon glue** — async `request`, safe-tokens set, proposal write + async
+   poll, allow/block decision; pass `safe_tokens` into the WebSocket path.
+4. **On-match policy** — `dlp.outbound_on_match` through manifest → render →
+   addon; `redact` surface scrub with fail-closed re-scan; policy dispatch in
+   the addon's outbound handler.
+5. **Tests + docs** — core/supervise/TUI/manifest/render unit tests; README
+   egress + supervisor notes.
+
+### Out of scope
+
+The deferrals enumerated under **Non-goals** — restart persistence, inbound /
+WebSocket-frame supervision, cross-encoding generalisation, replacing
+`dlp.outbound_detectors`, and making `redact` the default.
+
+## Proposed Design
+
+### New services / components
+
+A new proposal tool constant `egress-token-allow` (`TOOL_EGRESS_TOKEN_ALLOW`)
+is added to `supervise.TOOLS`, and the egress addon gains an in-memory
+safe-tokens set plus the policy-dispatch path that drives it.
+
+On an outbound block the addon dispatches on the resolved policy:
+
+- **Structural blocks always 403.** A `ScanResult` with no `matched` value
+  (CRLF injection) is a hard `403` regardless of policy — there is nothing to
+  redact or safelist.
+- **`redact`** runs `redact_tokens` over the body, non-`host` header values,
+  and path/query, then re-scans. If the re-scan is clean the (rewritten)
+  request is forwarded; if a block-severity match remains (e.g. in the
+  hostname, or a unicode-evasion token redaction can't reach) it fails closed
+  with a `403`.
+- **`block`** writes the `403` immediately.
+- **`supervise`** runs the queue-and-wait loop, falling back to `block` when
+  supervise isn't wired for the bottle.
+
+For `supervise`, the addon writes the proposal directly to
+`SUPERVISE_QUEUE_DIR` (the queue is bind-mounted into the sidecar bundle and
+shared by every daemon, exactly as git-gate's `gitleaks-allow` proposal in PRD
+0061 does). The proposal's `proposed_file` is a human-readable text payload
+built by `build_token_allow_payload`:
+
+```
+egress blocked an outbound request carrying a detected token
+host: api.example.com
+method: POST
+path: /v1/ingest
+detector: OpenAI API key found in body
+context: ...before ******** after...
+```
+
+The justification tells the operator to approve only if the value is a false
+positive or a credential the request legitimately needs. The addon then polls
+`<proposal-id>.response.json` for `EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS` (default
+300). `approved`/`modified` allow the request and add the value to the
+safe-tokens set; `rejected`, malformed responses, and timeout fail the request
+closed. The proposal + response are archived to `processed/` after a decision.
+Because the wait happens inside mitmproxy's asyncio loop, the addon's `request`
+hook is async and polls with `asyncio.sleep`, so concurrent flows are
+unaffected.
+
+### Existing code touched
+
+- **Policy threading.** `dlp.outbound_on_match` is a per-route enum threaded
+  from the bottle manifest (`manifest_egress`) through the resolved route
+  (`egress.EgressRoute`), the rendered `routes.yaml` (`egress_render_routes`),
+  and the addon's `Route` (`egress_addon_core`). Unset renders nothing and
+  resolves to `supervise` at request time. The `list-egress-routes`
+  introspection endpoint round-trips it so the agent's proposals preserve it.
+- **Provider-route default.** Agent-provider routes (the agent talking to its
+  own LLM API — `api.anthropic.com`, the Codex backend, etc.) are the worst
+  source of token-shaped false positives because the whole conversation payload
+  flows through them. `egress_routes_for_bottle` fills `outbound_on_match=redact`
+  on any provider route that doesn't set it explicitly; a provider that sets the
+  policy keeps its choice, and manifest routes are unaffected (they default to
+  `supervise`).
+- **Scanners.** `scan_outbound` (and the token detectors `scan_token_patterns`
+  / `scan_known_secrets` it calls) accept a `safe_tokens` set. A match whose
+  value is in `safe_tokens` is skipped, so an approved token no longer blocks;
+  the scanners keep searching past a safelisted match so a second, un-approved
+  secret in the same request is still caught. The WebSocket path is passed the
+  same `safe_tokens` set.
+- **Supervisor UI.** `cli/supervise.py` renders `egress-token-allow` like
+  `gitleaks-allow`: the text payload is shown, modify is unavailable (there is
+  no file patch to edit), and approval prompts for a non-empty reason recorded
+  in the response notes. There is no on-disk config diff, so — like
+  `gitleaks-allow` and `capability-block` — it writes no egress audit-log entry.
+- **Failure handling.** If `SUPERVISE_QUEUE_DIR` / `SUPERVISE_BOTTLE_SLUG` are
+  unset (supervise disabled for the bottle), the addon skips the queue and
+  returns the existing `403`. Any error writing the proposal or reading the
+  response also fails closed.
+
+### Data model changes
+
+- New per-route manifest field `dlp.outbound_on_match: block | redact |
+  supervise`, rendered into `routes.yaml` (omitted when unset).
+- `ScanResult` gains a `matched: str = ""` field carrying the raw substring the
+  detector matched. The token detectors populate it; the structural CRLF
+  detector leaves it empty. The value stays inside the egress sidecar process —
+  never written to a log line (logs use the redacted `context`) nor to the
+  proposal file.
+- Proposal text payload (above) plus `<proposal-id>.response.json` in
+  `SUPERVISE_QUEUE_DIR`, archived to `processed/` after a decision.
+- New env var `EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS` (default 300).
+
+### External dependencies
+
+None. Reuses the existing supervisor queue (`SUPERVISE_QUEUE_DIR`) and the
+mitmproxy addon framework already in the egress sidecar.
+
+## Open questions
+
+- Should `known_secrets` (provisioned `EGRESS_TOKEN_*` exfiltration) be
+  override-able at all, or only `token_patterns`? This PRD allows both —
+  approval is an explicit operator decision and the safe-tokens set matches the
+  exact found value — but a future revision could restrict `known_secrets` to
+  reject-only.
+
+## References
+
+- Issue #261
+- PRD 0061 — `gitleaks-allow` supervisor proposal pattern this reuses.

docs/research/local-vs-remote-agent-execution.md

@@ -22,7 +22,7 @@ escapes**, and **whether credentials are short-lived and scoped**.
 - Outbound: Docker containers have full internet access by default; no egress monitoring on most home networks
 - Lateral movement: compromised container can reach the LAN — NAS, other machines, internal services
 - Notable: CVE-2025-59536 (CVSS 8.7, Feb 2026) — a poisoned `.claude/settings.json` in a repo gives RCE when Claude Code opens it. `--dangerously-skip-permissions` removes the last gate.
-- Supply chain: MCP servers, skills, and npm packages pulled during agent execution. ~20% of ClawHub skills were found malicious in early 2026.
+- Supply chain: MCP servers, skills, and npm packages pulled during agent execution. A Jan 2026 large-scale empirical study of a 98,380-skill snapshot confirmed 157 malicious skills, ~71% of them credential harvesters. Exfiltration was overwhelmingly naive — plaintext HTTP to hardcoded endpoints; under 10% used any code obfuscation, and concealment was mostly at the documentation level, not the code level. ([Malicious Agent Skills in the Wild](https://arxiv.org/html/2602.06547v1), arXiv:2602.06547)
 
 **What local topology protects:**
 - No inbound attack surface — nothing listening on a public port

examples/bottles/claude.md

@@ -1,14 +1,14 @@
 ---
 agent_provider:
   template: claude
-
+  # auth_token names the host env var holding the Claude OAuth token. The
-egress:
+  # provider injects a provider-owned api.anthropic.com egress route that
-  routes:
+  # re-injects this token as the Bearer header; the agent only ever sees a
-    - host: api.anthropic.com
+  # placeholder CLAUDE_CODE_OAUTH_TOKEN. DLP defaults (token_patterns,
-      role: claude_code_oauth
+  # known_secrets outbound; naive_injection_detection inbound) apply to
-      auth:
+  # that route. To scan additional hosts, declare them under egress.routes
-        scheme: Bearer
+  # with per-route matches/dlp (see README "Egress route fields").
-        token_ref: BOT_BOTTLE_CLAUDE_OAUTH_TOKEN
+  auth_token: BOT_BOTTLE_CLAUDE_OAUTH_TOKEN
 ---
 
 Common Claude provider boundary. Drop this file into

tests/unit/test_agent_provider.py

@@ -168,6 +168,34 @@ class TestAgentProviderRuntime(unittest.TestCase):
         self.assertEqual("~/.claude/statusline.sh", settings["statusLine"]["command"])
         self.assertEqual("custom:bot-bottle-research-ui", settings["theme"])
 
+    def test_claude_plan_uses_startup_args_from_provider_settings(self):
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            plan = build_agent_provision_plan(
+                template="claude",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                provider_settings={
+                    "startup_args": ["--model", "opus"],
+                },
+            )
+        self.assertEqual(("--model", "opus"), plan.startup_args)
+
+    def test_codex_plan_uses_startup_args_from_provider_settings(self):
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            plan = build_agent_provision_plan(
+                template="codex",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                provider_settings={
+                    "startup_args": ["--model", "gpt-5-codex"],
+                },
+            )
+        self.assertEqual(("--model", "gpt-5-codex"), plan.startup_args)
+
     def test_codex_forward_host_credentials_populates_egress_routes(self):
         with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
             home = Path(tmp) / "host-codex"
@@ -394,6 +422,24 @@ class TestAgentProviderRuntime(unittest.TestCase):
         self.assertNotIn("OPENROUTER_API_KEY", plan.guest_env)
         self.assertTrue(provider["compat"]["supportsReasoningEffort"])
 
+    def test_pi_plan_appends_startup_args_from_provider_settings(self):
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            plan = build_agent_provision_plan(
+                template="pi",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                provider_settings={
+                    "models": ["qwen3:14b"],
+                    "startup_args": ["--no-stream"],
+                },
+            )
+        self.assertEqual(
+            ("--models", "ollama/qwen3:14b", "--no-stream"),
+            plan.startup_args,
+        )
+
     def test_pi_prompt_mode_appends_system_prompt_interactively(self):
         self.assertEqual(
             ["--append-system-prompt", "/home/node/.bot-bottle-prompt.txt"],

tests/unit/test_cli_start_settle.py

@@ -102,6 +102,27 @@ class TestAttachAgent(unittest.TestCase):
             bottle.argv,
         )
 
+    def test_remote_control_is_provider_startup_arg(self):
+        class Bottle:
+            argv: list[str] = []
+
+            def exec_agent(self, argv: list[str], *, tty: bool = True) -> int:
+                self.argv = list(argv)
+                return 0
+
+        bottle = Bottle()
+        exit_code = start_mod.attach_agent(
+            bottle,  # type: ignore[arg-type]
+            agent_provider_template="codex",
+            startup_args=("remote-control",),
+        )
+
+        self.assertEqual(0, exit_code)
+        self.assertEqual(
+            ["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+            bottle.argv,
+        )
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_contrib_codex_provider.py

@@ -29,6 +29,9 @@ from bot_bottle.supervise import SupervisePlan
 
 
 _URL = "http://supervise:9100/"
+_CODEX_DOCKERFILE = (
+    Path(__file__).resolve().parents[2] / "bot_bottle/contrib/codex/Dockerfile"
+)
 
 
 def _make_bottle(exec_result: ExecResult | None = None) -> MagicMock:
@@ -276,6 +279,12 @@ class TestCodexProvision(unittest.TestCase):
             )
 
 
+class TestCodexDockerfile(unittest.TestCase):
+    def test_installs_procps_for_remote_control_pid_management(self):
+        dockerfile = _CODEX_DOCKERFILE.read_text()
+        self.assertIn("procps", dockerfile)
+
+
 class TestCodexSuperviseMcp(unittest.TestCase):
     def test_noop_when_supervise_disabled(self):
         bottle = _make_bottle()

tests/unit/test_contrib_gitea_deploy_key.py

@@ -12,6 +12,7 @@ from bot_bottle.contrib.gitea.deploy_key_provisioner import (
     GiteaDeployKeyProvisioner,
     _split_owner_repo,
 )
+from bot_bottle.deploy_key_provisioner import DeployKeyCollisionError
 
 
 def _provisioner() -> GiteaDeployKeyProvisioner:
@@ -100,6 +101,30 @@ class TestCreate(unittest.TestCase):
                 provisioner.create("owner/repo", "title")
         self.assertIn("403", str(ctx.exception))
 
+    def test_create_raises_collision_error_on_422(self):
+        provisioner = _provisioner()
+        collision_body = json.dumps({
+            "errors": ["Key content already exists on this repository"],
+            "message": "422 Unprocessable Entity",
+        })
+        with patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.subprocess.run"
+        ), patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.urllib.request.urlopen",
+            side_effect=_http_error(422, collision_body),
+        ), patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_bytes",
+            return_value=b"pk",
+        ), patch(
+            "bot_bottle.contrib.gitea.deploy_key_provisioner.Path.read_text",
+            return_value="ssh-ed25519 AAAA\n",
+        ):
+            with self.assertRaises(DeployKeyCollisionError) as ctx:
+                provisioner.create("owner/repo", "my-title")
+        msg = str(ctx.exception)
+        self.assertIn("owner/repo", msg)
+        self.assertIn("my-title", msg)
+
 
 class TestDelete(unittest.TestCase):
     def test_delete_calls_correct_endpoint(self):

tests/unit/test_dlp_detectors.py

@@ -445,5 +445,63 @@ class TestKnownSecretsNewVariants(unittest.TestCase):
         self.assertIsNotNone(result)
 
 
+class TestMatchedAndSafeTokens(unittest.TestCase):
+    """PRD 0062: detectors carry the raw matched value, and a safelisted
+    value is skipped so the supervisor can approve a specific token."""
+
+    def test_token_pattern_sets_matched(self):
+        token = "ghp_" + "A" * 36
+        result = scan_token_patterns(f"token: {token}")
+        assert result is not None
+        self.assertEqual(token, result.matched)
+
+    def test_safe_token_is_skipped(self):
+        token = "ghp_" + "A" * 36
+        self.assertIsNone(
+            scan_token_patterns(f"token: {token}", safe_tokens={token})
+        )
+
+    def test_safe_token_does_not_mask_other_token(self):
+        safe = "ghp_" + "A" * 36
+        other = "AKIAIOSFODNN7EXAMPLE"
+        result = scan_token_patterns(
+            f"a={safe} b={other}", safe_tokens={safe},
+        )
+        assert result is not None
+        self.assertEqual(other, result.matched)
+        self.assertIn("AWS", result.reason)
+
+    def test_known_secret_sets_matched_and_safelist_skips(self):
+        secret = "supersecretvalue123"
+        env = {"EGRESS_TOKEN_FOO": secret}
+        result = scan_known_secrets(f"x={secret}", env=env)
+        assert result is not None
+        self.assertEqual(secret, result.matched)
+        self.assertIsNone(
+            scan_known_secrets(f"x={secret}", env=env, safe_tokens={secret})
+        )
+
+    def test_crlf_block_has_no_matched_value(self):
+        result = scan_crlf_injection("path%0d%0aHost: evil")
+        assert result is not None
+        self.assertEqual("", result.matched)
+
+
+class TestStripCrlf(unittest.TestCase):
+    def test_removes_url_encoded_crlf(self):
+        from bot_bottle.dlp_detectors import strip_crlf
+        out = strip_crlf("next=%0d%0aX-Injected: evil")
+        self.assertNotRegex(out, r"%0[dD]%0[aA]")
+
+    def test_removes_literal_header_injection(self):
+        from bot_bottle.dlp_detectors import strip_crlf
+        out = strip_crlf("value\r\nX-Injected: evil")
+        self.assertIsNone(scan_crlf_injection(out))
+
+    def test_leaves_clean_text_unchanged(self):
+        from bot_bottle.dlp_detectors import strip_crlf
+        self.assertEqual("/api/v1/data?q=hello", strip_crlf("/api/v1/data?q=hello"))
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_docker_bottle.py

@@ -136,6 +136,16 @@ class TestClaudeArgv(unittest.TestCase):
             argv,
         )
 
+    def test_codex_remote_control_startup_arg_does_not_receive_initial_prompt(self):
+        argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
+            ["--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+        )
+        self.assertEqual(
+            ["docker", "exec", "-it", "bot-bottle-dev-abc", "codex",
+             "--dangerously-bypass-approvals-and-sandbox", "remote-control"],
+            argv,
+        )
+
     def test_codex_resume_does_not_append_initial_prompt(self):
         argv = _codex_bottle("/home/node/.bot-bottle-prompt.txt").agent_argv(
             ["--dangerously-bypass-approvals-and-sandbox", "resume", "--last"],

tests/unit/test_docker_provision_git_user.py

@@ -31,7 +31,6 @@ class _Provider(AgentProvider):
         return AgentProviderRuntime(
             template="test", command="test", image="",
             prompt_mode="append_file", bypass_args=(), resume_args=(),
-            remote_control_args=(),
         )
     def provision_plan(self, **kwargs):  # type: ignore[override]
         raise NotImplementedError

tests/unit/test_egress.py

@@ -202,6 +202,23 @@ class TestProviderRouteMerge(unittest.TestCase):
         self.assertEqual((), routes[0].matches)
         self.assertEqual({}, egress_token_env_map(routes))
 
+    def test_provider_route_defaults_to_redact_on_match(self):
+        b = _bottle([])
+        pr = EgressRoute(host="api.anthropic.com")
+        routes = egress_routes_for_bottle(b, (pr,))
+        self.assertEqual("redact", routes[0].outbound_on_match)
+
+    def test_provider_route_explicit_on_match_preserved(self):
+        b = _bottle([])
+        pr = EgressRoute(host="api.anthropic.com", outbound_on_match="supervise")
+        routes = egress_routes_for_bottle(b, (pr,))
+        self.assertEqual("supervise", routes[0].outbound_on_match)
+
+    def test_manifest_route_does_not_get_redact_default(self):
+        b = _bottle([{"host": "api.example.com"}])
+        routes = egress_routes_for_bottle(b)
+        self.assertEqual("", routes[0].outbound_on_match)
+
     def test_two_provider_routes_with_same_token_ref_share_slot(self):
         b = _bottle([])
         routes = egress_routes_for_bottle(b, (
@@ -329,6 +346,23 @@ class TestRenderRoutes(unittest.TestCase):
         self.assertEqual(("token_patterns",), addon_routes[0].outbound_detectors)
         self.assertEqual((), addon_routes[0].inbound_detectors)
 
+    def test_outbound_on_match_round_trips(self):
+        from bot_bottle.egress_addon_core import load_routes
+        b = _bottle([{"host": "logs.example", "dlp": {
+            "outbound_on_match": "redact",
+        }}])
+        routes = egress_routes_for_bottle(b)
+        rendered = egress_render_routes(routes)
+        self.assertIn('outbound_on_match: "redact"', rendered)
+        addon_routes = load_routes(rendered)
+        self.assertEqual("redact", addon_routes[0].outbound_on_match)
+
+    def test_outbound_on_match_default_omitted_from_render(self):
+        b = _bottle([{"host": "x.example"}])
+        routes = egress_routes_for_bottle(b)
+        rendered = egress_render_routes(routes)
+        self.assertNotIn("outbound_on_match", rendered)
+
     def test_git_fetch_policy_round_trips(self):
         from bot_bottle.egress_addon_core import load_routes
         b = _bottle([{"host": "github.com", "git": {"fetch": True}}])

tests/unit/test_egress_addon_core.py

@@ -22,8 +22,10 @@ from bot_bottle.egress_addon_core import (
     MatchEntry,
     PathMatch,
     Route,
+    ScanResult,
     build_inbound_scan_text,
     build_outbound_scan_text,
+    build_token_allow_payload,
     decide,
     decide_git_fetch,
     evaluate_matches,
@@ -267,6 +269,25 @@ class TestParseDlp(unittest.TestCase):
                 "dlp": {"wat": True},
             }]})
 
+    def test_outbound_on_match_default_empty(self):
+        routes = parse_routes({"routes": [{"host": "x.example"}]})
+        self.assertEqual("", routes[0].outbound_on_match)
+
+    def test_outbound_on_match_parsed(self):
+        for policy in ("block", "redact", "supervise"):
+            routes = parse_routes({"routes": [{
+                "host": "x.example",
+                "dlp": {"outbound_on_match": policy},
+            }]})
+            self.assertEqual(policy, routes[0].outbound_on_match)
+
+    def test_outbound_on_match_invalid_rejected(self):
+        with self.assertRaises(ValueError):
+            parse_routes({"routes": [{
+                "host": "x.example",
+                "dlp": {"outbound_on_match": "nope"},
+            }]})
+
 
 # --- load_routes ---------------------------------------------------------
 
@@ -1167,5 +1188,92 @@ class TestScanInbound(unittest.TestCase):
         self.assertEqual("block", result.severity)
 
 
+class TestScanOutboundSafeTokens(unittest.TestCase):
+    """PRD 0062: scan_outbound threads the supervisor-approved safe-tokens
+    set into the token detectors."""
+
+    def test_safe_token_allows_request(self):
+        text = build_outbound_scan_text(
+            host="api.example.com", path="/v1/data", query="",
+            headers={}, body=f"key={_AWS_KEY}",
+        )
+        self.assertIsNone(
+            scan_outbound(_ROUTE, text, {}, safe_tokens={_AWS_KEY})
+        )
+
+    def test_unrelated_safe_token_still_blocks(self):
+        text = build_outbound_scan_text(
+            host="api.example.com", path="/v1/data", query="",
+            headers={}, body=f"key={_AWS_KEY}",
+        )
+        result = scan_outbound(_ROUTE, text, {}, safe_tokens={"ghp_" + "A" * 36})
+        self.assertIsNotNone(result)
+        assert result is not None
+        self.assertEqual(_AWS_KEY, result.matched)
+
+
+class TestScanOutboundCrlfText(unittest.TestCase):
+    """PRD 0062: CRLF is scanned only over the request line + headers
+    (crlf_text), never the body — a body is not an injection vector."""
+
+    def test_body_crlf_not_flagged_when_crlf_text_excludes_body(self):
+        # A form-encoded multi-line body legitimately contains %0d%0a.
+        body = "comment=line1%0d%0aline2"
+        full = build_outbound_scan_text(
+            host="api.example.com", path="/submit", query="",
+            headers={}, body=body,
+        )
+        crlf_text = build_outbound_scan_text(
+            host="api.example.com", path="/submit", query="",
+            headers={}, body="",
+        )
+        self.assertIsNone(scan_outbound(_ROUTE, full, {}, crlf_text=crlf_text))
+
+    def test_request_line_crlf_still_flagged(self):
+        full = build_outbound_scan_text(
+            host="api.example.com", path="/p", query="next=%0d%0aX:evil",
+            headers={}, body="",
+        )
+        crlf_text = full
+        result = scan_outbound(_ROUTE, full, {}, crlf_text=crlf_text)
+        self.assertIsNotNone(result)
+        assert result is not None
+        self.assertEqual("block", result.severity)
+
+    def test_default_crlf_text_scans_full_blob(self):
+        # Backward compatibility: crlf_text=None scans everything (body too).
+        full = build_outbound_scan_text(
+            host="api.example.com", path="/submit", query="",
+            headers={}, body="x=%0d%0aX:evil",
+        )
+        self.assertIsNotNone(scan_outbound(_ROUTE, full, {}))
+
+
+class TestBuildTokenAllowPayload(unittest.TestCase):
+    def test_payload_includes_context_and_no_raw_token(self):
+        result = ScanResult(
+            severity="block",
+            reason="AWS access key found in body",
+            location="body",
+            context="key=******** tail",
+            matched=_AWS_KEY,
+        )
+        payload = build_token_allow_payload(
+            "api.example.com", "POST", "/v1/ingest", result,
+        )
+        self.assertIn("host: api.example.com", payload)
+        self.assertIn("method: POST", payload)
+        self.assertIn("path: /v1/ingest", payload)
+        self.assertIn("AWS access key found in body", payload)
+        self.assertIn("key=******** tail", payload)
+        # The raw matched value must never appear in the proposal file.
+        self.assertNotIn(_AWS_KEY, payload)
+
+    def test_payload_omits_context_line_when_empty(self):
+        result = ScanResult(severity="block", reason="r", matched="x")
+        payload = build_token_allow_payload("h", "GET", "/", result)
+        self.assertNotIn("context:", payload)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_git_gate.py

@@ -199,6 +199,30 @@ class TestHookRender(unittest.TestCase):
         self.assertIn('set -- "$@" --push-option="$opt"', hook)
         self.assertIn('git push "$@" origin "$refspec"', hook)
 
+    def test_inline_gitleaks_allow_routes_to_supervisor(self):
+        hook = git_gate_render_hook()
+        # First gitleaks runs normally; only if that passes does the
+        # hook ask gitleaks to ignore inline allow comments and report
+        # the suppressed findings for human approval.
+        self.assertIn("--ignore-gitleaks-allow", hook)
+        self.assertIn("--report-format=json", hook)
+        self.assertIn('"tool": "gitleaks-allow"', hook)
+        self.assertIn("SUPERVISE_QUEUE_DIR", hook)
+        self.assertIn("SUPERVISE_BOTTLE_SLUG", hook)
+        self.assertIn("supervisor approved # gitleaks:allow", hook)
+        self.assertIn("supervisor rejected # gitleaks:allow", hook)
+
+    def test_inline_gitleaks_allow_fails_closed_without_supervisor(self):
+        hook = git_gate_render_hook()
+        self.assertIn(
+            "cannot route # gitleaks:allow finding to supervisor; refusing push",
+            hook,
+        )
+        self.assertIn(
+            "supervisor approval timed out for # gitleaks:allow; refusing push",
+            hook,
+        )
+
 
 class TestAccessHookRender(unittest.TestCase):
     def test_access_hook_refreshes_origin_on_upload_pack(self):

tests/unit/test_log.py

@@ -0,0 +1,127 @@
+"""Unit: leveled + structured logging wrappers (issue #252).
+
+Locks three properties of bot_bottle.log:
+  - backward compatibility — default output is byte-identical to the
+    original bare wrappers, so the 100+ existing single-string call
+    sites are unaffected;
+  - context rendering — an optional mapping becomes a parseable
+    ` [k=v ...]` suffix;
+  - level gating — BOT_BOTTLE_LOG_LEVEL filters by severity, debug is
+    silent by default, and error always surfaces.
+"""
+
+from __future__ import annotations
+
+import contextlib
+import io
+import unittest
+from typing import Callable
+from unittest import mock
+
+from bot_bottle import log
+
+
+def _capture(
+    fn: Callable[..., None],
+    *args: object,
+    env: dict[str, str] | None = None,
+    **kwargs: object,
+) -> str:
+    buf = io.StringIO()
+    patched = mock.patch.dict("os.environ", env or {}, clear=False)
+    with patched, contextlib.redirect_stderr(buf):
+        fn(*args, **kwargs)
+    return buf.getvalue()
+
+
+class TestBackwardCompat(unittest.TestCase):
+    """No context + default level → exactly the legacy lines."""
+
+    def test_info(self):
+        self.assertEqual("bot-bottle: hello\n", _capture(log.info, "hello"))
+
+    def test_warn(self):
+        self.assertEqual(
+            "bot-bottle: warning: careful\n", _capture(log.warn, "careful")
+        )
+
+    def test_error(self):
+        self.assertEqual(
+            "bot-bottle: error: boom\n", _capture(log.error, "boom")
+        )
+
+
+class TestContext(unittest.TestCase):
+    def test_appends_sorted_parseable_suffix(self):
+        out = _capture(
+            log.error, "rpc failed", context={"slug": "abc123", "code": "-32603"}
+        )
+        # keys sorted: code before slug
+        self.assertEqual(
+            "bot-bottle: error: rpc failed [code=-32603 slug=abc123]\n", out
+        )
+
+    def test_quotes_values_with_whitespace(self):
+        out = _capture(
+            log.info, "did thing", context={"path": "/a b/c", "ok": "yes"}
+        )
+        self.assertEqual(
+            'bot-bottle: did thing [ok=yes path="/a b/c"]\n', out
+        )
+
+    def test_empty_context_is_noop_suffix(self):
+        self.assertEqual(
+            "bot-bottle: x\n", _capture(log.info, "x", context={})
+        )
+
+
+class TestLevels(unittest.TestCase):
+    def test_debug_silent_by_default(self):
+        self.assertEqual("", _capture(log.debug, "trace"))
+
+    def test_debug_emits_when_level_lowered(self):
+        out = _capture(log.debug, "trace", env={"BOT_BOTTLE_LOG_LEVEL": "debug"})
+        self.assertEqual("bot-bottle: debug: trace\n", out)
+
+    def test_error_level_suppresses_info_and_warn(self):
+        env = {"BOT_BOTTLE_LOG_LEVEL": "error"}
+        self.assertEqual("", _capture(log.info, "i", env=env))
+        self.assertEqual("", _capture(log.warn, "w", env=env))
+        # error still surfaces — nothing sits above it
+        self.assertEqual(
+            "bot-bottle: error: e\n", _capture(log.error, "e", env=env)
+        )
+
+    def test_unknown_level_falls_back_to_default(self):
+        # garbage value → default INFO threshold, so info still prints
+        out = _capture(log.info, "i", env={"BOT_BOTTLE_LOG_LEVEL": "loud"})
+        self.assertEqual("bot-bottle: i\n", out)
+
+    def test_warning_alias_accepted(self):
+        env = {"BOT_BOTTLE_LOG_LEVEL": "warning"}
+        self.assertEqual("", _capture(log.info, "i", env=env))
+        self.assertEqual(
+            "bot-bottle: warning: w\n", _capture(log.warn, "w", env=env)
+        )
+
+
+class TestDie(unittest.TestCase):
+    def test_die_still_raises_and_prints_error(self):
+        buf = io.StringIO()
+        with contextlib.redirect_stderr(buf):
+            with self.assertRaises(log.Die) as cm:
+                log.die("fatal thing")
+        self.assertEqual("fatal thing", cm.exception.message)
+        self.assertIn("bot-bottle: error: fatal thing", buf.getvalue())
+
+    def test_die_surfaces_even_at_error_level(self):
+        buf = io.StringIO()
+        with mock.patch.dict("os.environ", {"BOT_BOTTLE_LOG_LEVEL": "error"}):
+            with contextlib.redirect_stderr(buf):
+                with self.assertRaises(log.Die):
+                    log.die("still fatal")
+        self.assertIn("bot-bottle: error: still fatal", buf.getvalue())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_manifest_egress.py

@@ -167,13 +167,40 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
                 },
             })
 
-    def test_settings_rejected_for_claude(self):
+    def test_startup_args_allowed_for_claude(self):
+        b = _provider_config_bottle({
+            "template": "claude",
+            "settings": {"startup_args": ["--model", "opus"]},
+        })
+        self.assertEqual(
+            {"startup_args": ["--model", "opus"]},
+            b.agent_provider.settings,
+        )
+
+    def test_startup_args_allowed_for_codex(self):
+        b = _provider_config_bottle({
+            "template": "codex",
+            "settings": {"startup_args": ["--model", "gpt-5-codex"]},
+        })
+        self.assertEqual(
+            {"startup_args": ["--model", "gpt-5-codex"]},
+            b.agent_provider.settings,
+        )
+
+    def test_provider_specific_settings_still_rejected_for_claude(self):
         with self.assertRaises(ManifestError):
             _provider_config_bottle({
                 "template": "claude",
                 "settings": {"models": ["qwen2.5-coder:7b"]},
             })
 
+    def test_startup_args_must_be_string_array(self):
+        with self.assertRaises(ManifestError):
+            _provider_config_bottle({
+                "template": "codex",
+                "settings": {"startup_args": ["--model", 42]},
+            })
+
     def test_settings_models_must_be_non_empty_string_array(self):
         with self.assertRaises(ManifestError):
             _provider_config_bottle({
@@ -302,6 +329,24 @@ class TestDlp(unittest.TestCase):
                 "bogus": True,
             }}])
 
+    def test_outbound_on_match_omitted_is_empty(self):
+        b = _bottle([{"host": "x.example"}])
+        self.assertEqual("", b.egress.routes[0].OutboundOnMatch)
+
+    def test_outbound_on_match_accepts_policies(self):
+        for policy in ("block", "redact", "supervise"):
+            with self.subTest(policy=policy):
+                b = _bottle([{"host": "x.example", "dlp": {
+                    "outbound_on_match": policy,
+                }}])
+                self.assertEqual(policy, b.egress.routes[0].OutboundOnMatch)
+
+    def test_outbound_on_match_rejects_unknown_value(self):
+        with self.assertRaises(ManifestError):
+            _bottle([{"host": "x.example", "dlp": {
+                "outbound_on_match": "allow",
+            }}])
+
 
 class TestGitPolicy(unittest.TestCase):
     def test_omitted_means_https_git_fetch_disabled(self):

tests/unit/test_plan_print_parity.py

@@ -130,7 +130,7 @@ def _capture_print(plan: DockerBottlePlan | SmolmachinesBottlePlan) -> list[str]
     orig = sys.stderr
     sys.stderr = buf
     try:
-        plan.print(remote_control=False)
+        plan.print()
     finally:
         sys.stderr = orig
     return buf.getvalue().splitlines()

tests/unit/test_smolmachines_provision.py

@@ -44,7 +44,6 @@ class _Provider(AgentProvider):
         return AgentProviderRuntime(
             template="test", command="test", image="",
             prompt_mode="append_file", bypass_args=(), resume_args=(),
-            remote_control_args=(),
         )
     def provision_plan(self, **kwargs):  # type: ignore[override]
         raise NotImplementedError

tests/unit/test_supervise.py

@@ -17,6 +17,7 @@ from bot_bottle.supervise import (
     STATUS_MODIFIED,
     STATUS_REJECTED,
     TOOL_CAPABILITY_BLOCK,
+    TOOL_GITLEAKS_ALLOW,
     archive_proposal,
     audit_log_path,
     list_pending_proposals,
@@ -317,18 +318,30 @@ class TestToolConstants(unittest.TestCase):
     def test_tools_tuple_matches_individual_constants(self):
         self.assertEqual(
             (
-                supervise.TOOL_ALLOW,
+                supervise.TOOL_EGRESS_ALLOW,
                 TOOL_CAPABILITY_BLOCK,
                 supervise.TOOL_EGRESS_BLOCK,
+                TOOL_GITLEAKS_ALLOW,
+                supervise.TOOL_EGRESS_TOKEN_ALLOW,
                 supervise.TOOL_LIST_EGRESS_ROUTES,
             ),
             supervise.TOOLS,
         )
 
+    def test_token_allow_proposal_roundtrips(self):
+        p = Proposal.new(
+            bottle_slug="dev",
+            tool=supervise.TOOL_EGRESS_TOKEN_ALLOW,
+            proposed_file="host: api.example.com\n",
+            justification="false positive",
+            current_file_hash="h",
+        )
+        self.assertEqual(p, Proposal.from_dict(p.to_dict()))
+
     def test_component_map_has_egress_entries(self):
         self.assertEqual(
             {
-                supervise.TOOL_ALLOW: "egress",
+                supervise.TOOL_EGRESS_ALLOW: "egress",
                 supervise.TOOL_EGRESS_BLOCK: "egress",
             },
             supervise.COMPONENT_FOR_TOOL,

tests/unit/test_supervise_cli.py

@@ -19,6 +19,8 @@ from bot_bottle.supervise import (
     STATUS_MODIFIED,
     STATUS_REJECTED,
     TOOL_CAPABILITY_BLOCK,
+    TOOL_GITLEAKS_ALLOW,
+    TOOL_EGRESS_TOKEN_ALLOW,
     read_audit_entries,
     read_response,
     sha256_hex,
@@ -31,8 +33,10 @@ FIXED = datetime(2026, 5, 25, 12, 0, 0, tzinfo=timezone.utc)
 def _proposal(slug: str = "dev", tool: str = TOOL_CAPABILITY_BLOCK) -> Proposal:
     payloads = {
         TOOL_CAPABILITY_BLOCK: "FROM python:3.13\n",
-        supervise.TOOL_ALLOW: "routes:\n  - host: example.com\n",
+        supervise.TOOL_EGRESS_ALLOW: "routes:\n  - host: example.com\n",
         supervise.TOOL_EGRESS_BLOCK: "routes:\n  - host: example.com\n",
+        TOOL_GITLEAKS_ALLOW: "file: tests/test_fixture.py\nline: 3\n",
+        TOOL_EGRESS_TOKEN_ALLOW: "host: api.example.com\ndetector: token\n",
     }
     payload = payloads.get(tool, "")
     return Proposal.new(
@@ -170,6 +174,63 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
         self.assertEqual(STATUS_APPROVED, entries[0].operator_action)
         self.assertEqual("needed for dev", entries[0].justification)
 
+    def test_approve_gitleaks_allow_leaves_response_for_gate(self):
+        qp = self._enqueue(tool=TOOL_GITLEAKS_ALLOW)
+        supervise_cli.approve(qp, notes="dummy fixture")
+        # Gate polls the queue dir for the response; TUI must not archive it.
+        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.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 queue dir for the response; the TUI must
+        # not archive it (the addon archives after reading).
+        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)
+        supervise_cli.approve(qp, notes="false positive")
+        self.assertEqual([], read_audit_entries("egress", "dev"))
+
+    def test_tui_token_allow_requires_reason(self):
+        qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_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_token_allow_writes_reason(self):
+        qp = self._enqueue(tool=TOOL_EGRESS_TOKEN_ALLOW)
+        with patch.object(supervise_cli, "_prompt", return_value="legit"):
+            status = supervise_cli._approve_from_tui(None, qp)  # type: ignore[arg-type]
+        self.assertIn("approved egress-token-allow", status)
+        resp = read_response(qp.queue_dir, qp.proposal.id)
+        self.assertEqual("legit", resp.notes)
+
+    def test_suffix_for_token_allow_is_txt(self):
+        self.assertEqual(".txt", supervise_cli._suffix_for_tool(TOOL_EGRESS_TOKEN_ALLOW))
+
 
 # class TestCapabilityApplyWiring(_FakeHomeMixin, unittest.TestCase):
 #     # DISABLED — capability_apply functionality is currently commented out.

tests/unit/test_supervise_server.py

@@ -59,7 +59,7 @@ class TestValidation(unittest.TestCase):
 
     def test_egress_routes_yaml_is_validated(self):
         validate_proposed_file(
-            _sv.TOOL_ALLOW,
+            _sv.TOOL_EGRESS_ALLOW,
             "routes:\n  - host: example.com\n",
         )
 
@@ -147,7 +147,7 @@ class TestHandleToolsList(unittest.TestCase):
         names = [t["name"] for t in result["tools"]]  # type: ignore[index]
         self.assertEqual(
             sorted([
-                _sv.TOOL_ALLOW,
+                _sv.TOOL_EGRESS_ALLOW,
                 _sv.TOOL_CAPABILITY_BLOCK,
                 _sv.TOOL_EGRESS_BLOCK,
                 _sv.TOOL_LIST_EGRESS_ROUTES,
@@ -181,7 +181,7 @@ class TestHandleToolsList(unittest.TestCase):
         self.assertNotIn("required", schema)  # type: ignore[operator]
 
     def test_egress_tools_take_routes_yaml_and_justification(self):
-        for tool_name in (_sv.TOOL_ALLOW, _sv.TOOL_EGRESS_BLOCK):
+        for tool_name in (_sv.TOOL_EGRESS_ALLOW, _sv.TOOL_EGRESS_BLOCK):
             with self.subTest(tool_name=tool_name):
                 tool = next(t for t in TOOL_DEFINITIONS if t["name"] == tool_name)
                 schema = tool["inputSchema"]
@@ -244,7 +244,7 @@ class TestHandleToolsCall(unittest.TestCase):
         try:
             result = handle_tools_call(
                 {
-                    "name": _sv.TOOL_ALLOW,
+                    "name": _sv.TOOL_EGRESS_ALLOW,
                     "arguments": {
                         "routes_yaml": "routes:\n  - host: example.com\n",
                         "justification": "need example.com",
@@ -451,7 +451,7 @@ class TestHttpEndToEnd(unittest.TestCase):
         self.assertEqual(1, result["id"])
         names = [t["name"] for t in result["result"]["tools"]]  # type: ignore[index]
         self.assertIn(_sv.TOOL_CAPABILITY_BLOCK, names)
-        self.assertIn(_sv.TOOL_ALLOW, names)
+        self.assertIn(_sv.TOOL_EGRESS_ALLOW, names)
         self.assertIn(_sv.TOOL_EGRESS_BLOCK, names)
 
     def test_unknown_method_returns_jsonrpc_error(self):