fix(egress): strip injected Authorization and redact bodies in LOG_FULL path

_log_request and _log_response wrote headers and bodies to stderr verbatim. _log_request also included the sidecar-injected upstream Authorization value, exposing live bearer tokens on every allowed request under LOG_FULL. Apply redact_tokens to all header values and bodies in both log functions; exclude the authorization header from _log_request entirely since its value is always a live sidecar-injected credential by the time _log_request runs. Closes #257
feat(provider): support startup args settings
2026-06-24 23:04:22 -04:00 · 2026-06-24 22:51:27 -04:00 · 2026-06-25 02:23:12 +00:00 · 2026-06-25 01:23:10 +00:00 · 2026-06-24 21:10:31 -04:00 · 2026-06-24 20:40:36 -04:00
44 changed files with 1713 additions and 596 deletions
@@ -8,40 +8,26 @@
 [![pylint](https://img.shields.io/badge/pylint-9.93%2F10-brightgreen)](https://github.com/PyCQA/pylint)
 [![pyright](https://img.shields.io/badge/pyright-0%20errors-brightgreen)](https://github.com/microsoft/pyright)

-**Run any coding agent like it might be compromised — and lose nothing when it is.**
+**Problem:** Developer wants to run a coding agent without supervision, but they don't want a prompt injected or misbehaving agent wrecking their environment or exfiltrating sensitive data.

-bot-bottle is a provider-neutral, security-first substrate for autonomous agents. Bring Claude Code, Codex, or your own harness; each one runs in an ephemeral, per-agent "bottle" it cannot modify, where every byte of egress is scanned for exfiltration and capabilities are narrowed to exactly what the task declares.
+**Solution:** Ephemeral, per agent "bottles" the agent cannot modify that scan all traffic for data exfiltration and limit capabilities and egress to only what the agent needs.

-**Problem:** You want to let a coding agent run unsupervised, but a prompt-injected or misbehaving agent — or a poisoned repo, MCP server, or skill — can wreck your environment or exfiltrate your secrets. Locking yourself to one vendor's cloud doesn't fix that; it just moves the blast radius.
-
-**Solution:** A neutral control plane that runs *whatever agent you choose* inside an isolation boundary the agent can't touch: TLS-bumped egress allowlisting, outbound/inbound DLP, gitleaks-gated pushes, and host secrets the agent never sees. Swap the agent; keep the guarantees.
-
-## Why bot-bottle
-
-### A neutral substrate — bring your own agent
-
- **Provider-agnostic by design** — Claude and Codex ship built in; any other agent (Gemini, Aider, a local-model wrapper) is a drop-in plugin at `~/.bot-bottle/contrib/<name>/` — no fork, no PR against this repo. The manifest accepts any provider template, and the isolation, egress, and git guarantees are identical across all of them.
- **One control plane, every harness** — the same bottle, egress policy, and supervise flow wrap whichever agent you run, so switching or mixing providers doesn't change your security posture.
- **Composable bottles (`extends:`)** — keep provider/runtime policy in one base bottle (e.g. `claude.md`) and overlay task bottles on top.
-
-### An isolation boundary the agent can't touch
+## Features

 - **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.
 - **Trust boundary at `$HOME`** — bottles (credentials, egress, remotes) live only under `~/.bot-bottle/bottles/`. Repos may ship agents but not bottles, so a cloned repo can't redirect an env var to an attacker host.
-
-### Isolation that matches your host
-
+- **Composable bottles (`extends:`)** — keep provider/runtime policy in one base bottle (e.g. `claude.md`) and overlay task bottles on top.
 - **Parallel, isolated bottles** — each bottle runs in its own backend-owned isolation boundary; bottles don't share state or talk to each other.
+- **Provider templates (Claude, Codex)** — `Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile. Claude auth via long-lived OAuth token; Codex via opt-in host device-auth forwarding.
 - **gVisor auto-detect** — on Linux hosts where `runsc` is registered with Docker, every bottle launches under it for a userspace syscall barrier; no manifest config required.
 - **Apple Container backend (macOS default when available)** — runs the agent and sidecar bundle with Apple's `container` CLI, using a host-only agent network plus a separate sidecar egress network.
 - **Smolmachines backend** — runs the agent in a libkrun micro-VM while the sidecar bundle stays in Docker. TSI and smolmachines DNS filtering close the raw DNS exfiltration gap that exists in the legacy Docker backend.
 - **Legacy Docker backend** — still available for examples, CI, and hosts without Apple Container via `BOT_BOTTLE_BACKEND=docker` or `--backend=docker`.

-Per-provider auth (Claude long-lived OAuth token; Codex opt-in host device-auth forwarding) and per-provider images (`Dockerfile.claude` / `Dockerfile.codex`, or a bottle-supplied Dockerfile) are configured on the bottle — see [Manifest](#manifest).
-
 ## Architecture

 On the default macOS Apple Container backend, a bottle is an agent container on a host-only internal network plus a sidecar bundle attached to both that internal network and a NAT egress network. The agent gets HTTP(S)_PROXY and CA bundle env vars pointing at the sidecar's internal-network IP, so HTTP/HTTPS traffic flows through the sidecar instead of direct egress. `bottle.git` / git-gate is intentionally deferred on this backend until a safe Apple Container key-delivery path exists.
@@ -83,27 +69,6 @@ The Docker topology looks like this:

 When the agent exits, `cli.py` tears down every sidecar and both networks; nothing about a bottle persists between runs.

-## Install
-
-Install the CLI with the bootstrap script:
-
-```sh
-curl -fsSL https://gitea.dideric.is/didericis/bot-bottle/raw/branch/main/install.sh | sh
-```
-
-The script checks Python 3.11+, checks Docker daemon reachability, creates the `~/.bot-bottle/` config directories, installs the Python package with `pipx` when available or `pip --user` otherwise, then runs:
-
-```sh
-bot-bottle doctor
-```
-
-Python-native installers can use the package metadata directly:
-
-```sh
-pipx install git+https://gitea.dideric.is/didericis/bot-bottle.git
-uv tool install git+https://gitea.dideric.is/didericis/bot-bottle.git
-```
-
 ## Quickstart

 On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The smolmachines backend requires Docker on the host for the sidecar bundle plus smolvm. The legacy Docker backend requires Docker. Claude bottles also need a long-lived Claude Code OAuth token (`claude setup-token`) exported as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`.
@@ -184,8 +149,11 @@ You help maintain Gitea-hosted projects.
 | `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
@@ -1,96 +0,0 @@
-# Per-bottle sidecar bundle image (PRD 0024).
-#
-# Collapses the prior per-sidecar images (egress, git-gate,
-# supervise) into one. A small stdlib-Python init supervisor at
-# /app/sidecar_init.py spawns all daemons, forwards SIGTERM, and
-# propagates per-daemon stdout/stderr to the container log with a
-# `[name]` prefix. See PRD 0024 for the rationale.
-#
-# Layout:
-#
-#   /usr/bin/gitleaks                    gitleaks binary
-#   /app/egress_addon.py + siblings      mitmproxy addon (egress)
-#   /app/egress-entrypoint.sh            mitmdump launcher
-#   /app/supervise_server.py + .py       supervise MCP server
-#   /app/sidecar_init.py                 PID 1 supervisor
-#   /etc/egress/routes.yaml              bind-mounted at run time
-#   /etc/git-gate/pre-receive            docker-cp'd at start time
-#   /git-gate-entrypoint.sh              docker-cp'd at start time
-#   /git-gate/creds/*                    docker-cp'd at start time
-#   /git/*                               bare repos, populated at runtime
-#   /run/supervise/queue/                bind-mounted at run time
-#   /home/mitmproxy/.mitmproxy/          mitmproxy CA dir
-#
-# Exposed ports inside the container:
-#   9099  egress (mitmproxy, agent-facing HTTPS proxy)
-#   9418  git-gate (git-daemon)
-#   9420  git-gate smart HTTP (smolmachines agent-facing transport)
-#   9100  supervise (MCP HTTP)
-
-# Stage 1: gitleaks binary. The upstream gitleaks image is alpine
-# with the binary at /usr/bin/gitleaks. Pinned by digest in lockstep
-# with Dockerfile.git-gate's prior base (now deleted at chunk 3).
-FROM zricethezav/gitleaks@sha256:c00b6bd0aeb3071cbcb79009cb16a60dd9e0a7c60e2be9ab65d25e6bc8abbb7f AS gitleaks-src
-
-# Stage 2: assembly. mitmproxy/mitmproxy is debian-slim-based with
-# Python + mitmdump pre-installed — heavier than the others, so
-# this stage starts there and pulls the standalone binaries in.
-FROM mitmproxy/mitmproxy:11.1.3
-
-# Run as root inside the bundle. The bundle is the isolation
-# boundary; per-daemon user separation inside it is not load-bearing
-# and complicates the supervisor's spawn path.
-USER root
-
-# Runtime system deps:
-#   git supplies the `git daemon` subcommand (no separate package)
-#     plus the core `git` binary the pre-receive hook invokes.
-#   openssh-client supplies the upstream SSH transport the
-#     pre-receive hook uses to forward accepted refs.
-#   ca-certificates is needed for mitmdump upstream TLS (the
-#     base image already has it; listed for explicitness).
-RUN apt-get update \
- && apt-get install -y --no-install-recommends \
-      git openssh-client ca-certificates \
- && rm -rf /var/lib/apt/lists/*
-
-# Pull the standalone binaries into the final image.
-COPY --from=gitleaks-src /usr/bin/gitleaks /usr/bin/gitleaks
-
-# Project Python: addon + server modules + the init supervisor.
-# Kept flat under /app/ so mitmdump's loader resolves them as
-# top-level siblings (absolute imports), matching the prior
-# Dockerfile.egress / Dockerfile.supervise layout.
-COPY bot_bottle/egress_addon_core.py /app/egress_addon_core.py
-COPY bot_bottle/egress_addon.py      /app/egress_addon.py
-COPY bot_bottle/dlp_detectors.py     /app/dlp_detectors.py
-COPY bot_bottle/yaml_subset.py       /app/yaml_subset.py
-COPY bot_bottle/supervise.py         /app/supervise.py
-COPY bot_bottle/supervise_server.py  /app/supervise_server.py
-COPY bot_bottle/sidecar_init.py      /app/sidecar_init.py
-COPY bot_bottle/git_http_backend.py  /app/git_http_backend.py
-COPY bot_bottle/egress_entrypoint.sh /app/egress-entrypoint.sh
-RUN chmod +x /app/egress-entrypoint.sh
-
-# Pre-create runtime directories the compose renderer + start
-# step expect to exist. `docker cp` does not create intermediate
-# dirs, and bind mounts won't either if the parent is missing.
-RUN mkdir -p \
-      /etc/egress \
-      /etc/git-gate \
-      /git-gate/creds \
-      /git \
-      /run/supervise/queue \
-      /home/mitmproxy/.mitmproxy
-
-# Documentation only — the compose renderer publishes whichever
-# subset the bottle uses.
-EXPOSE 8888 9099 9418 9420 9100
-
-# WORKDIR matches Dockerfile.supervise's prior layout so the
-# in-app same-dir import in supervise_server.py stays deterministic.
-WORKDIR /app
-
-# PID 1 is the supervisor. It owns signal handling and exit-code
-# propagation; no `exec` chain in the entrypoint itself.
-ENTRYPOINT ["python3", "/app/sidecar_init.py"]
@@ -371,6 +371,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,
@@ -58,17 +58,10 @@ from .sidecar_bundle import (
 )


-# Repo root or installed site-packages root, used as the build context for
-# Dockerfiles that COPY bot_bottle source files.
+# Repo root, used as the build context for the bundle Dockerfile.
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)


-def _sidecar_bundle_dockerfile() -> str:
-    if (Path(_REPO_DIR) / SIDECAR_BUNDLE_DOCKERFILE).is_file():
-        return SIDECAR_BUNDLE_DOCKERFILE
-    return f"bot_bottle/{SIDECAR_BUNDLE_DOCKERFILE}"
-
-
 def bottle_plan_to_compose(plan: DockerBottlePlan) -> dict[str, Any]:
    """Render a Compose v2 spec dict from a fully-resolved
    DockerBottlePlan.
@@ -190,7 +183,7 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
        "image": SIDECAR_BUNDLE_IMAGE,
        "build": {
            "context": _REPO_DIR,
-            "dockerfile": _sidecar_bundle_dockerfile(),
+            "dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
        },
        "container_name": sidecar_bundle_container_name(plan.slug),
        "networks": {
@@ -12,10 +12,9 @@ from __future__ import annotations
 import os


-# Bundle image. Defaults to a built-locally tag. Source checkouts
-# build from the repo-root Dockerfile.sidecars; installed packages
-# build from the packaged copy under bot_bottle/.
-# Operators pinning to a published digest can override via env.
+# Bundle image. Defaults to a built-locally tag (built from the
+# repo's Dockerfile.sidecars via compose `build:`). Operators
+# pinning to a published digest can override via env.
 SIDECAR_BUNDLE_IMAGE = os.environ.get(
    "BOT_BOTTLE_SIDECAR_IMAGE",
    "bot-bottle-sidecars:latest",
@@ -1,6 +1,6 @@
 """Main CLI dispatcher.

-Commands: cleanup, commit, doctor, edit, info, init, list, resume, start, supervise
+Commands: cleanup, commit, edit, info, init, list, resume, start, supervise
 """

 from __future__ import annotations
@@ -13,7 +13,6 @@ from ._common import PROG
 from . import list as _list_mod
 from .cleanup import cmd_cleanup
 from .commit import cmd_commit
-from .doctor import cmd_doctor
 from .edit import cmd_edit
 from .info import cmd_info
 from .init import cmd_init
@@ -26,7 +25,6 @@ cmd_list = _list_mod.cmd_list
 COMMANDS = {
    "cleanup": cmd_cleanup,
    "commit": cmd_commit,
-    "doctor": cmd_doctor,
    "edit": cmd_edit,
    "info": cmd_info,
    "init": cmd_init,
@@ -42,7 +40,6 @@ def usage() -> None:
    sys.stderr.write("Commands:\n")
    sys.stderr.write("  cleanup   stop and remove all active bot-bottle containers\n")
    sys.stderr.write("  commit    snapshot a running bottle's container state to a Docker image\n")
-    sys.stderr.write("  doctor    check Python, Docker, and bot-bottle config prerequisites\n")
    sys.stderr.write("  edit      open an agent in vim for editing\n")
    sys.stderr.write("  info      print env, skills, and prompt details for a named agent\n")
    sys.stderr.write("  init      interactively create a new agent and add it to bot-bottle.json\n")
@@ -6,7 +6,7 @@ import os
 import sys
 from pathlib import Path

-PROG = Path(sys.argv[0]).name or "bot-bottle"
+PROG = "cli.py"
 USER_CWD = os.getcwd()
 REPO_DIR = str(Path(__file__).resolve().parent.parent.parent)

@@ -1,73 +0,0 @@
-"""doctor: validate host prerequisites for running bot-bottle."""
-
-from __future__ import annotations
-
-import argparse
-import shutil
-import subprocess
-import sys
-from pathlib import Path
-
-from ._common import PROG
-
-
-def _ok(label: str, detail: str) -> None:
-    print(f"ok: {label}: {detail}")
-
-
-def _fail(label: str, detail: str) -> None:
-    print(f"fail: {label}: {detail}")
-
-
-def _check_python() -> bool:
-    version = sys.version_info
-    detail = f"{version.major}.{version.minor}.{version.micro}"
-    if version >= (3, 11):
-        _ok("python", detail)
-        return True
-    _fail("python", f"{detail}; need 3.11 or newer")
-    return False
-
-
-def _check_docker() -> bool:
-    docker = shutil.which("docker")
-    if not docker:
-        _fail("docker", "docker command not found")
-        return False
-    try:
-        result = subprocess.run(
-            [docker, "info"],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
-            timeout=10,
-        )
-    except (OSError, subprocess.TimeoutExpired) as exc:
-        _fail("docker", f"daemon check failed: {exc}")
-        return False
-    if result.returncode == 0:
-        _ok("docker", "daemon reachable")
-        return True
-    _fail("docker", "daemon not reachable")
-    return False
-
-
-def _check_config_dir() -> bool:
-    config = Path.home() / ".bot-bottle"
-    if config.is_dir():
-        _ok("config", str(config))
-        return True
-    _fail("config", f"{config} does not exist")
-    return False
-
-
-def cmd_doctor(argv: list[str]) -> int:
-    parser = argparse.ArgumentParser(prog=f"{PROG} doctor", add_help=True)
-    parser.parse_args(argv)
-
-    checks = (
-        _check_python(),
-        _check_docker(),
-        _check_config_dir(),
-    )
-    return 0 if all(checks) else 1
@@ -51,9 +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,
@@ -65,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:
@@ -139,9 +145,9 @@ 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 == TOOL_GITLEAKS_ALLOW:
+    if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
        return ".txt"
    return ".txt"

@@ -171,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,
@@ -212,8 +218,8 @@ def _approve_from_tui(
    notes: str = "",
 ) -> str:
    """Approve from curses, prompting for any tool-specific audit note."""
-    if qp.proposal.tool == TOOL_GITLEAKS_ALLOW and final_file is None:
-        notes = _prompt(stdscr, "allow reason (test fixture/false positive): ")
+    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)
@@ -292,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
@@ -408,8 +417,8 @@ def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore
            except ApplyError as e:
                status_line = f"apply failed: {e}"
        elif key == ord("m"):
-            if qp.proposal.tool == TOOL_GITLEAKS_ALLOW:
-                status_line = "modify unavailable for gitleaks-allow"
+            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:
@@ -522,7 +531,7 @@ def _detail_view(
                pass
            return
        elif key == ord("m"):
-            if qp.proposal.tool == TOOL_GITLEAKS_ALLOW:
+            if qp.proposal.tool in _REPORT_ONLY_TOOLS:
                return
            edited = _modify(stdscr, qp)
            if edited is not None:
@@ -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
@@ -115,8 +116,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 +201,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,
@@ -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
@@ -79,8 +80,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 +165,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),
@@ -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}"
@@ -21,6 +21,7 @@ from ...agent_provider import (
    AgentProvisionDir,
    AgentProvisionFile,
    AgentProvisionPlan,
+    provider_startup_args,
 )
 from ...egress import EgressRoute
 from ...log import die, info
@@ -199,6 +200,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 +221,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)),),
@@ -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."""

@@ -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)
-        if m is not None:
+        for m in pattern.finditer(normalized):
+            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",
 ]
@@ -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"


@@ -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")
@@ -121,31 +160,42 @@ class EgressAddon:
        )

    def _log_request(self, flow: http.HTTPFlow) -> None:
+        headers = {
+            k: redact_tokens(v, env=os.environ)
+            for k, v in flow.request.headers.items()
+            if k.lower() != "authorization"
+        }
+        body = redact_tokens(flow.request.get_text(strict=False) or "", env=os.environ)
        sys.stderr.write(
            json.dumps({
                "event": "egress_request",
                "host": redact_tokens(flow.request.pretty_host, env=os.environ),
                "method": flow.request.method,
                "path": redact_tokens(flow.request.path, env=os.environ),
-                "headers": dict(flow.request.headers),
-                "body": flow.request.get_text(strict=False) or "",
+                "headers": headers,
+                "body": body,
            })
            + "\n"
        )

    def _log_response(self, flow: http.HTTPFlow) -> None:
+        headers = {
+            k: redact_tokens(v, env=os.environ)
+            for k, v in flow.response.headers.items()
+        }
+        body = redact_tokens(flow.response.get_text(strict=False) or "", env=os.environ)
        sys.stderr.write(
            json.dumps({
                "event": "egress_response",
                "host": flow.request.pretty_host,
                "status": flow.response.status_code,
-                "headers": dict(flow.response.headers),
-                "body": flow.response.get_text(strict=False) or "",
+                "headers": headers,
+                "body": body,
            })
            + "\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 +207,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 ""
-            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)
+            if not await self._handle_outbound_dlp(flow, route):
                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 +261,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 +508,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 +527,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()]
@@ -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
-    # by outbound_detectors config.
-    result = scan_crlf_injection(text)
+    # CRLF injection is only an attack in the request line + headers, never the
+    # body: an HTTP body is delimited by Content-Length, so CRLF bytes there
+    # 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",
@@ -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:
-    print(f"bot-bottle: {msg}", file=sys.stderr)
+def _threshold() -> int:
+    """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:
-    print(f"bot-bottle: warning: {msg}", file=sys.stderr)
+def _format_context(context: Mapping[str, object] | None) -> str:
+    """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:
-    print(f"bot-bottle: error: {msg}", file=sys.stderr)
+def _emit(
+    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:
-    error(msg)
+def die(msg: str, *, context: Mapping[str, object] | None = None) -> NoReturn:
+    error(msg, context=context)
    raise Die(1, msg)
@@ -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):
@@ -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})
@@ -50,14 +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,
 )

@@ -76,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",
 }

@@ -555,7 +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",
@@ -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:
@@ -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.
@@ -1,75 +0,0 @@
-# PRD prd-new: Install script
-
- **Status:** Active
- **Author:** didericis
- **Created:** 2026-06-06
- **Issue:** #197
-
-## Summary
-
-Add a proper Python package distribution and a thin `install.sh` bootstrapper so users can install bot-bottle with a single command without cloning the repo.
-
-## Problem
-
-There is currently no install path for new users. The only way to run bot-bottle is to clone the repo and invoke `cli.py` directly. This blocks any HN-style public demo: readers want `curl | sh` or `pipx install`, not a manual clone-and-configure flow.
-
-## Goals / Success Criteria
-
- `curl -fsSL <url>/install.sh | sh` (or equivalent) leaves a working `bot-bottle` command on PATH.
- Python-native users can install with `pipx install bot-bottle` or `uv tool install bot-bottle`.
- `install.sh` validates prerequisites (Python ≥ 3.11, Docker) and exits with a clear message if they are missing. It does not silently install Docker.
- `install.sh` runs `bot-bottle doctor` (or equivalent diagnostic) after install to confirm the environment is ready.
- The package has no runtime pip dependencies (stdlib-only, matching the existing constraint).
-
-## Non-goals
-
- Bundling a Python runtime or producing a standalone binary.
- Automatic Docker installation.
- Plugin architecture changes (out of scope; see issue #197 for future direction).
- Publishing to PyPI in this PR — the package structure is the deliverable; publishing is a separate step.
-
-## Design
-
-### Package structure
-
-Add a minimal `pyproject.toml` at the repo root:
-
-```toml
-[project]
-name = "bot-bottle"
-version = "0.1.0"
-requires-python = ">=3.11"
-dependencies = []
-
-[project.scripts]
-bot-bottle = "bot_bottle.cli:main"
-```
-
-The existing `bot_bottle/` package and `cli.py` entry point already contain the logic; this just wires up the standard entry point. `cli.py` may need a small refactor to expose a `main()` callable if it uses `if __name__ == "__main__"` only.
-
-### `install.sh`
-
-A thin bootstrapper that:
-
-1. Checks `python3 --version` ≥ 3.11; exits with instructions if not met.
-2. Checks `docker info` exits 0; exits with instructions if Docker is not running.
-3. Installs via `pipx` if available, otherwise falls back to `pip install --user`.
-4. Runs `bot-bottle doctor` to verify the install.
-
-The script must be idempotent (safe to re-run) and must not require `sudo`.
-
-### `bot-bottle doctor`
-
-A new subcommand that checks and reports:
-
- Python version.
- Docker daemon reachability.
- Whether `~/.bot-bottle/` config directory exists.
-
-Exits 0 if all checks pass, non-zero otherwise.
-
-## Decisions
-
- `install.sh` is hosted from the repo's raw Gitea URL for now:
-  `https://gitea.dideric.is/didericis/bot-bottle/raw/branch/main/install.sh`.
- Should `version` in `pyproject.toml` be driven by a git tag at build time (e.g. via `hatch-vcs`) or kept as a static string? Static is simpler for now.
@@ -0,0 +1,85 @@
+# PRD prd-new: LOG_FULL egress logging credential redaction
+
+- **Status:** Draft
+- **Author:** claude
+- **Created:** 2026-06-25
+- **Issue:** #257
+
+## Summary
+
+The `LOG_FULL` egress logging path (`_log_request` and `_log_response` in `egress_addon.py`) writes request/response headers and bodies to stderr without redaction and includes the sidecar-injected upstream `Authorization` header verbatim. This PR applies `redact_tokens` to header values and bodies in both log functions and strips the injected `Authorization` header from request logs entirely.
+
+## Problem
+
+`LOG_FULL` (log level 2) is intended for debugging egress traffic. When active it calls `_log_request` and `_log_response`. Both functions have two related bugs:
+
+1. **Injected `Authorization` header exposure.** `_log_request` is called *after* the sidecar injects upstream credentials (`flow.request.headers["authorization"] = decision.inject_authorization`). The full header dict — including the live credential — is serialized to stderr. Any log collector that ingests the egress container's stderr will receive the upstream bearer token in plaintext.
+
+2. **Unredacted bodies and header values.** Neither `_log_request` nor `_log_response` passes body or header values through `redact_tokens`. By contrast, `_req_ctx` (used for block/warn events) already calls `redact_tokens` on path and host. Any provisioned secret or recognized token pattern that appears in a request body, response body, or non-Authorization header value will be logged verbatim under `LOG_FULL`.
+
+These two bugs compose: an agent that enables `LOG_FULL` and simultaneously triggers a request that carries a known token gains a write path from credentials → egress logs.
+
+## Goals / Success Criteria
+
+- `_log_request` never logs the `authorization` header in any form.
+- `_log_request` applies `redact_tokens(value, env=os.environ)` to every other header value before serializing.
+- `_log_request` applies `redact_tokens(body, env=os.environ)` to the request body before logging.
+- `_log_response` applies `redact_tokens(value, env=os.environ)` to every response header value before logging.
+- `_log_response` applies `redact_tokens(body, env=os.environ)` to the response body before logging.
+- Unit tests cover each of the five cases above.
+
+## Non-goals
+
+- Redacting host or path in the full-log path (already covered by `_req_ctx` for block/warn events; `_log_request` already calls `redact_tokens` on host and path).
+- Suppressing `LOG_FULL` or adding a new log level.
+- Changing the outbound DLP scan logic.
+
+## Design
+
+### `_log_request`
+
+```python
+def _log_request(self, flow: http.HTTPFlow) -> None:
+    headers = {
+        k: redact_tokens(v, env=os.environ)
+        for k, v in flow.request.headers.items()
+        if k.lower() != "authorization"
+    }
+    body = redact_tokens(flow.request.get_text(strict=False) or "", env=os.environ)
+    sys.stderr.write(
+        json.dumps({
+            "event": "egress_request",
+            "host": redact_tokens(flow.request.pretty_host, env=os.environ),
+            "method": flow.request.method,
+            "path": redact_tokens(flow.request.path, env=os.environ),
+            "headers": headers,
+            "body": body,
+        })
+        + "\n"
+    )
+```
+
+The `authorization` key is excluded because by the time `_log_request` is called the sidecar has already injected the upstream credential (`decision.inject_authorization`). Logging it would write a live bearer token to stderr on every allowed request. There is no safe subset to log — the value is always a live credential or empty.
+
+### `_log_response`
+
+```python
+def _log_response(self, flow: http.HTTPFlow) -> None:
+    headers = {
+        k: redact_tokens(v, env=os.environ)
+        for k, v in flow.response.headers.items()
+    }
+    body = redact_tokens(flow.response.get_text(strict=False) or "", env=os.environ)
+    sys.stderr.write(
+        json.dumps({
+            "event": "egress_response",
+            "host": flow.request.pretty_host,
+            "status": flow.response.status_code,
+            "headers": headers,
+            "body": body,
+        })
+        + "\n"
+    )
+```
+
+Response headers don't carry injected credentials, so no header name is suppressed — only the values are scrubbed by `redact_tokens`.
@@ -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
@@ -1,50 +0,0 @@
-#!/bin/sh
-set -eu
-
-PACKAGE_SPEC="${BOT_BOTTLE_INSTALL_SPEC:-git+https://gitea.dideric.is/didericis/bot-bottle.git}"
-MIN_PYTHON="3.11"
-
-say() {
-    printf 'bot-bottle install: %s\n' "$*" >&2
-}
-
-die() {
-    say "error: $*"
-    exit 1
-}
-
-command -v python3 >/dev/null 2>&1 || die "python3 is required (version ${MIN_PYTHON} or newer)"
-
-python3 - <<'PY' || die "python3 3.11 or newer is required"
-import sys
-
-raise SystemExit(0 if sys.version_info >= (3, 11) else 1)
-PY
-
-command -v docker >/dev/null 2>&1 || die "Docker is required; install Docker and start the daemon, then re-run this script"
-docker info >/dev/null 2>&1 || die "Docker is installed but the daemon is not reachable; start Docker and re-run this script"
-
-mkdir -p \
-    "${HOME}/.bot-bottle/agents" \
-    "${HOME}/.bot-bottle/bottles" \
-    "${HOME}/.bot-bottle/contrib"
-
-if command -v pipx >/dev/null 2>&1; then
-    say "installing with pipx"
-    pipx install --force "${PACKAGE_SPEC}"
-else
-    say "pipx not found; installing with python3 -m pip --user"
-    python3 -m pip install --user --upgrade "${PACKAGE_SPEC}"
-fi
-
-if command -v bot-bottle >/dev/null 2>&1; then
-    BOT_BOTTLE_BIN="bot-bottle"
-elif [ -x "${HOME}/.local/bin/bot-bottle" ]; then
-    BOT_BOTTLE_BIN="${HOME}/.local/bin/bot-bottle"
-    say "using ${BOT_BOTTLE_BIN}; add ${HOME}/.local/bin to PATH for future shells"
-else
-    die "bot-bottle was installed but is not on PATH"
-fi
-
-say "running bot-bottle doctor"
-"${BOT_BOTTLE_BIN}" doctor
@@ -1,27 +0,0 @@
-[build-system]
-requires = ["setuptools>=68"]
-build-backend = "setuptools.build_meta"
-
-[project]
-name = "bot-bottle"
-version = "0.1.0"
-description = "Self-hosted sandbox for AI coding agents with egress controls"
-readme = "README.md"
-requires-python = ">=3.11"
-license = { text = "Apache-2.0" }
-dependencies = []
-
-[project.scripts]
-bot-bottle = "bot_bottle.cli:main"
-
-[tool.setuptools.packages.find]
-include = ["bot_bottle*"]
-
-[tool.setuptools.package-data]
-bot_bottle = [
-    "Dockerfile.sidecars",
-    "egress_entrypoint.sh",
-    "contrib/claude/Dockerfile",
-    "contrib/codex/Dockerfile",
-    "contrib/pi/Dockerfile",
-]
@@ -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"],
@@ -1,51 +0,0 @@
-"""Unit: `bot-bottle doctor` host prerequisite checks."""
-
-from __future__ import annotations
-
-import tempfile
-import unittest
-from pathlib import Path
-from unittest.mock import MagicMock, patch
-
-from bot_bottle.cli import doctor
-
-
-class TestDoctor(unittest.TestCase):
-    def test_success_when_prerequisites_present(self):
-        with tempfile.TemporaryDirectory() as tmp, patch.object(
-            doctor.Path, "home", return_value=Path(tmp),
-        ), patch.object(
-            doctor.shutil, "which", return_value="/usr/bin/docker",
-        ), patch.object(
-            doctor.subprocess, "run",
-            return_value=MagicMock(returncode=0),
-        ):
-            Path(tmp, ".bot-bottle").mkdir()
-            self.assertEqual(0, doctor.cmd_doctor([]))
-
-    def test_missing_config_fails(self):
-        with tempfile.TemporaryDirectory() as tmp, patch.object(
-            doctor.Path, "home", return_value=Path(tmp),
-        ), patch.object(
-            doctor.shutil, "which", return_value="/usr/bin/docker",
-        ), patch.object(
-            doctor.subprocess, "run",
-            return_value=MagicMock(returncode=0),
-        ):
-            self.assertEqual(1, doctor.cmd_doctor([]))
-
-    def test_missing_docker_fails_before_daemon_check(self):
-        with tempfile.TemporaryDirectory() as tmp, patch.object(
-            doctor.Path, "home", return_value=Path(tmp),
-        ), patch.object(
-            doctor.shutil, "which", return_value=None,
-        ), patch.object(
-            doctor.subprocess, "run",
-        ) as run:
-            Path(tmp, ".bot-bottle").mkdir()
-            self.assertEqual(1, doctor.cmd_doctor([]))
-            run.assert_not_called()
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -301,19 +301,6 @@ class TestSidecarBundleShape(unittest.TestCase):
        self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
        self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])

-    def test_bundle_uses_packaged_dockerfile_when_root_missing(self):
-        from bot_bottle.backend.docker import compose as compose_mod
-
-        original = compose_mod._REPO_DIR
-        try:
-            compose_mod._REPO_DIR = "/tmp/does-not-exist"
-            self.assertEqual(
-                "bot_bottle/Dockerfile.sidecars",
-                compose_mod._sidecar_bundle_dockerfile(),
-            )
-        finally:
-            compose_mod._REPO_DIR = original
-
    def test_bundle_container_name_uses_sidecars_prefix(self):
        sc = self._render()["services"]["sidecars"]
        self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])
@@ -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):
@@ -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()
@@ -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}}])
@@ -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()
@@ -0,0 +1,273 @@
+"""Unit: LOG_FULL credential redaction in _log_request / _log_response (issue #257).
+
+egress_addon.py is sidecar-only code that depends on mitmproxy, which is
+not installed on the host. This file pre-populates sys.modules with the
+minimum mocks needed so EgressAddon can be imported and tested without the
+real mitmproxy package."""
+
+from __future__ import annotations
+
+import json
+import sys
+import types
+import unittest
+from io import StringIO
+from unittest.mock import patch
+
+
+# ---------------------------------------------------------------------------
+# Sidecar-import shims — must run before importing egress_addon
+# ---------------------------------------------------------------------------
+
+def _ensure_shims() -> None:
+    if "mitmproxy" not in sys.modules:
+        _mm = types.ModuleType("mitmproxy")
+        _mh = types.ModuleType("mitmproxy.http")
+        _mm.http = _mh
+        sys.modules["mitmproxy"] = _mm
+        sys.modules["mitmproxy.http"] = _mh
+    if "egress_addon_core" not in sys.modules:
+        import bot_bottle.egress_addon_core as _core
+        sys.modules["egress_addon_core"] = _core
+
+
+_ensure_shims()
+
+from bot_bottle.egress_addon import EgressAddon  # noqa: E402  (import after shims)
+from bot_bottle.egress_addon_core import Config, LOG_FULL  # noqa: E402
+
+
+# ---------------------------------------------------------------------------
+# Helpers
+# ---------------------------------------------------------------------------
+
+def _addon() -> EgressAddon:
+    """Return a bare EgressAddon with LOG_FULL config and no routes file."""
+    a: EgressAddon = EgressAddon.__new__(EgressAddon)
+    a.config = Config(routes=(), log=LOG_FULL)
+    a.safe_tokens = set()
+    a._supervise_queue_dir = ""
+    a._supervise_slug = ""
+    a._token_allow_timeout = 300.0
+    return a
+
+
+class _Headers:
+    def __init__(self, d: dict[str, str]) -> None:
+        self._d = d
+
+    def items(self) -> list[tuple[str, str]]:
+        return list(self._d.items())
+
+
+class _Request:
+    def __init__(
+        self,
+        host: str = "api.example.com",
+        method: str = "POST",
+        path: str = "/v1/messages",
+        headers: dict[str, str] | None = None,
+        body: str = "",
+    ) -> None:
+        self.pretty_host = host
+        self.method = method
+        self.path = path
+        self.headers = _Headers(headers or {})
+        self._body = body
+
+    def get_text(self, *, strict: bool = True) -> str:
+        return self._body
+
+
+class _Response:
+    def __init__(
+        self,
+        status_code: int = 200,
+        headers: dict[str, str] | None = None,
+        body: str = "",
+    ) -> None:
+        self.status_code = status_code
+        self.headers = _Headers(headers or {})
+        self._body = body
+
+    def get_text(self, *, strict: bool = True) -> str:
+        return self._body
+
+
+class _Flow:
+    def __init__(
+        self,
+        request: _Request | None = None,
+        response: _Response | None = None,
+    ) -> None:
+        self.request = request or _Request()
+        self.response = response or _Response()
+
+
+def _log_request(addon: EgressAddon, flow: _Flow) -> dict:
+    buf = StringIO()
+    with patch("sys.stderr", buf):
+        addon._log_request(flow)  # type: ignore[arg-type]
+    return json.loads(buf.getvalue())
+
+
+def _log_response(addon: EgressAddon, flow: _Flow) -> dict:
+    buf = StringIO()
+    with patch("sys.stderr", buf):
+        addon._log_response(flow)  # type: ignore[arg-type]
+    return json.loads(buf.getvalue())
+
+
+# ---------------------------------------------------------------------------
+# _log_request — authorization header stripped
+# ---------------------------------------------------------------------------
+
+
+class TestLogRequestAuthorizationStripped(unittest.TestCase):
+    def test_lowercase_authorization_excluded(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={"authorization": "Bearer sk-real-secret"}))
+        entry = _log_request(addon, flow)
+        self.assertNotIn("authorization", entry["headers"])
+
+    def test_titlecase_authorization_excluded(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={"Authorization": "Bearer sk-real-secret"}))
+        entry = _log_request(addon, flow)
+        self.assertNotIn("Authorization", entry["headers"])
+        self.assertNotIn("authorization", entry["headers"])
+
+    def test_non_auth_headers_retained(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={
+            "authorization": "Bearer sk-real-secret",
+            "content-type": "application/json",
+        }))
+        entry = _log_request(addon, flow)
+        self.assertIn("content-type", entry["headers"])
+        self.assertEqual("application/json", entry["headers"]["content-type"])
+
+    def test_no_authorization_header_logs_all_others(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={"x-request-id": "abc"}))
+        entry = _log_request(addon, flow)
+        self.assertEqual({"x-request-id": "abc"}, entry["headers"])
+
+
+# ---------------------------------------------------------------------------
+# _log_request — body redaction
+# ---------------------------------------------------------------------------
+
+
+_OPENAI_KEY = "sk-" + "A" * 48
+
+
+class TestLogRequestBodyRedacted(unittest.TestCase):
+    def test_token_pattern_in_body_scrubbed(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(body=f"key={_OPENAI_KEY}"))
+        entry = _log_request(addon, flow)
+        self.assertNotIn(_OPENAI_KEY, entry["body"])
+        self.assertIn("********", entry["body"])
+
+    def test_provisioned_secret_in_body_scrubbed(self) -> None:
+        addon = _addon()
+        secret = "provisioned-egress-secret-xyz"
+        flow = _Flow(request=_Request(body=f"token={secret}"))
+        with patch.dict("os.environ", {"EGRESS_TOKEN_0": secret}):
+            entry = _log_request(addon, flow)
+        self.assertNotIn(secret, entry["body"])
+        self.assertIn("********", entry["body"])
+
+    def test_clean_body_preserved(self) -> None:
+        addon = _addon()
+        payload = '{"model": "claude-3", "max_tokens": 1024}'
+        flow = _Flow(request=_Request(body=payload))
+        entry = _log_request(addon, flow)
+        self.assertEqual(payload, entry["body"])
+
+
+# ---------------------------------------------------------------------------
+# _log_request — non-authorization header value redaction
+# ---------------------------------------------------------------------------
+
+
+class TestLogRequestHeaderValuesRedacted(unittest.TestCase):
+    def test_token_in_custom_header_scrubbed(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={"x-api-key": _OPENAI_KEY}))
+        entry = _log_request(addon, flow)
+        self.assertNotIn(_OPENAI_KEY, entry["headers"].get("x-api-key", ""))
+        self.assertIn("********", entry["headers"].get("x-api-key", ""))
+
+    def test_clean_header_value_preserved(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(headers={"accept": "application/json"}))
+        entry = _log_request(addon, flow)
+        self.assertEqual("application/json", entry["headers"]["accept"])
+
+
+# ---------------------------------------------------------------------------
+# _log_response — body redaction
+# ---------------------------------------------------------------------------
+
+
+class TestLogResponseBodyRedacted(unittest.TestCase):
+    def test_token_pattern_in_response_body_scrubbed(self) -> None:
+        addon = _addon()
+        flow = _Flow(
+            request=_Request(),
+            response=_Response(body=f'{{"key": "{_OPENAI_KEY}"}}'),
+        )
+        entry = _log_response(addon, flow)
+        self.assertNotIn(_OPENAI_KEY, entry["body"])
+        self.assertIn("********", entry["body"])
+
+    def test_provisioned_secret_in_response_body_scrubbed(self) -> None:
+        addon = _addon()
+        secret = "provisioned-egress-secret-xyz"
+        flow = _Flow(
+            request=_Request(),
+            response=_Response(body=f'{{"token": "{secret}"}}'),
+        )
+        with patch.dict("os.environ", {"EGRESS_TOKEN_0": secret}):
+            entry = _log_response(addon, flow)
+        self.assertNotIn(secret, entry["body"])
+        self.assertIn("********", entry["body"])
+
+    def test_clean_response_body_preserved(self) -> None:
+        addon = _addon()
+        flow = _Flow(request=_Request(), response=_Response(body='{"result": "ok"}'))
+        entry = _log_response(addon, flow)
+        self.assertEqual('{"result": "ok"}', entry["body"])
+
+
+# ---------------------------------------------------------------------------
+# _log_response — response header value redaction
+# ---------------------------------------------------------------------------
+
+
+class TestLogResponseHeaderValuesRedacted(unittest.TestCase):
+    def test_token_in_response_header_scrubbed(self) -> None:
+        addon = _addon()
+        flow = _Flow(
+            request=_Request(),
+            response=_Response(headers={"set-cookie": f"token={_OPENAI_KEY}"}),
+        )
+        entry = _log_response(addon, flow)
+        cookie_val = entry["headers"].get("set-cookie", "")
+        self.assertNotIn(_OPENAI_KEY, cookie_val)
+        self.assertIn("********", cookie_val)
+
+    def test_clean_response_header_preserved(self) -> None:
+        addon = _addon()
+        flow = _Flow(
+            request=_Request(),
+            response=_Response(headers={"content-type": "application/json"}),
+        )
+        entry = _log_response(addon, flow)
+        self.assertEqual("application/json", entry["headers"]["content-type"])
+
+
+if __name__ == "__main__":
+    unittest.main()
@@ -1,34 +0,0 @@
-"""Unit: install.sh static contract checks."""
-
-from __future__ import annotations
-
-import subprocess
-import unittest
-from pathlib import Path
-
-
-ROOT = Path(__file__).resolve().parents[2]
-
-
-class TestInstallScript(unittest.TestCase):
-    def test_shell_syntax(self):
-        result = subprocess.run(
-            ["sh", "-n", str(ROOT / "install.sh")],
-            check=False,
-            capture_output=True,
-            text=True,
-        )
-        self.assertEqual("", result.stderr)
-        self.assertEqual(0, result.returncode)
-
-    def test_contract_phrases(self):
-        script = (ROOT / "install.sh").read_text(encoding="utf-8")
-        self.assertIn("python3", script)
-        self.assertIn("docker info", script)
-        self.assertIn("pipx install --force", script)
-        self.assertIn("pip install --user --upgrade", script)
-        self.assertIn('"${BOT_BOTTLE_BIN}" doctor', script)
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -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()
@@ -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):
@@ -1,27 +0,0 @@
-"""Unit: Python package metadata for install script PRD."""
-
-from __future__ import annotations
-
-import tomllib
-import unittest
-from pathlib import Path
-
-
-ROOT = Path(__file__).resolve().parents[2]
-
-
-class TestPyproject(unittest.TestCase):
-    def test_console_script_and_no_runtime_dependencies(self):
-        data = tomllib.loads((ROOT / "pyproject.toml").read_text(encoding="utf-8"))
-        project = data["project"]
-        self.assertEqual("bot-bottle", project["name"])
-        self.assertEqual(">=3.11", project["requires-python"])
-        self.assertEqual([], project["dependencies"])
-        self.assertEqual(
-            "bot_bottle.cli:main",
-            project["scripts"]["bot-bottle"],
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()
@@ -318,19 +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,
@@ -20,6 +20,7 @@ from bot_bottle.supervise import (
    STATUS_REJECTED,
    TOOL_CAPABILITY_BLOCK,
    TOOL_GITLEAKS_ALLOW,
+    TOOL_EGRESS_TOKEN_ALLOW,
    read_audit_entries,
    read_response,
    sha256_hex,
@@ -32,9 +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(
@@ -196,6 +198,39 @@ class TestApproveReject(_FakeHomeMixin, unittest.TestCase):
        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.
@@ -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):
Author	SHA1	Message	Date
didericis-claude	1f96619c6a	fix(egress): strip injected Authorization and redact bodies in LOG_FULL path lint / lint (push) Failing after 2m15s Details test / unit (pull_request) Successful in 43s Details test / integration (pull_request) Successful in 25s Details _log_request and _log_response wrote headers and bodies to stderr verbatim. _log_request also included the sidecar-injected upstream Authorization value, exposing live bearer tokens on every allowed request under LOG_FULL. Apply redact_tokens to all header values and bodies in both log functions; exclude the authorization header from _log_request entirely since its value is always a live sidecar-injected credential by the time _log_request runs. Closes #257	2026-06-24 23:04:22 -04:00
didericis	ecaae708f7	feat(provider): support startup args settings test / unit (pull_request) Successful in 41s Details test / integration (pull_request) Successful in 26s Details lint / lint (push) Successful in 2m12s Details test / unit (push) Successful in 41s Details test / integration (push) Successful in 26s Details Update Quality Badges / update-badges (push) Successful in 2m9s Details	2026-06-24 22:51:27 -04:00
didericis-claude	2e790268b0	fix(deploy-key): raise DeployKeyCollisionError on 422 key conflicts lint / lint (push) Successful in 2m7s Details test / unit (push) Successful in 46s Details test / integration (push) Successful in 25s Details Update Quality Badges / update-badges (push) Successful in 2m7s Details 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.	2026-06-25 02:23:12 +00:00
didericis-claude	a421d1d688	Rename TOOL_ALLOW to TOOL_EGRESS_ALLOW lint / lint (push) Successful in 3m50s Details test / unit (push) Successful in 38s Details test / integration (push) Successful in 23s Details Update Quality Badges / update-badges (push) Successful in 1m38s Details The constant and its MCP tool name ("allow" → "egress-allow") were the only supervise tools without an egress-scoped identifier, despite the tool being egress-only (routes.yaml payload, COMPONENT_FOR_TOOL maps it to "egress", always grouped with TOOL_EGRESS_BLOCK). The rename brings it in line with TOOL_EGRESS_BLOCK and TOOL_EGRESS_TOKEN_ALLOW, and adds TOOL_EGRESS_ALLOW and TOOL_EGRESS_BLOCK to __all__ (both were previously absent).	2026-06-25 01:23:10 +00:00
didericis	d2d50be65a	Restructure PRD 0062 to the init-prd template test / unit (pull_request) Successful in 39s Details test / integration (pull_request) Successful in 25s Details lint / lint (push) Successful in 2m8s Details test / unit (push) Successful in 41s Details test / integration (push) Successful in 25s Details Update Quality Badges / update-badges (push) Successful in 1m32s Details Conform the PRD to the standard PRD-new skeleton: add a Scope section (In scope / Out of scope), rename Design -> Proposed Design and split its prose into New services / Existing code touched / Data model changes / External dependencies, fold the old Implementation chunks into In scope, and add a References section. No change in substance. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com>	2026-06-24 21:10:31 -04:00
didericis	1ad710a041	Default agent-provider routes to the redact on-match policy lint / lint (push) Successful in 1m42s Details test / unit (pull_request) Successful in 34s Details test / integration (pull_request) Successful in 16s Details Provider routes (the agent talking to its own LLM API — api.anthropic.com, the Codex backend, etc.) carry the whole conversation payload, which is the worst source of token-shaped false positives. egress_routes_for_bottle now fills outbound_on_match=redact on any provider route that doesn't set it explicitly, so a match there is scrubbed and forwarded rather than blocked or queued for the operator. A provider that sets the policy keeps its choice; manifest routes still default to supervise. Tests: provider route gets redact default, explicit provider policy preserved, manifest route unaffected. README + PRD 0062 updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HnvBjPZC5V7qeQpFbQdDmS	2026-06-24 20:40:36 -04:00
didericis	b411577e76	Stop scanning the request body for CRLF injection lint / lint (push) Successful in 1m41s Details test / unit (pull_request) Successful in 31s Details test / integration (pull_request) Successful in 18s Details A 403 "egress DLP: URL-encoded CRLF (%0d%0a)" was firing on legitimate requests (e.g. the Claude Code login flow) and bypassing the on-match policy entirely, because CRLF blocks carry no matched value and were routed straight to a hard 403. Root cause: CRLF injection is only an attack in the request line and headers. An HTTP body is delimited by Content-Length, so CRLF bytes in the body cannot split the request — but the scan flattened the body into the same blob it checked, so form-encoded / multi-line body content (which legitimately contains %0d%0a) tripped it. Fix: - scan_outbound takes a crlf_text param; the addon scans CRLF only over the body-excluded request line + headers. crlf_text=None keeps the old full-blob behavior for host-side callers/tests; the websocket path passes "" since a data frame is not a request line. - The redact policy now also scrubs CRLF (new strip_crlf helper) from the path and headers, so redact is a complete escape hatch and structural CRLF in the URL/headers can be forwarded when a route opts into it. Tests: strip_crlf unit tests; scan_outbound crlf_text body-exclusion and backward-compat tests. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HnvBjPZC5V7qeQpFbQdDmS	2026-06-24 20:37:26 -04:00
didericis	cdfaaa3de8	Add dlp.outbound_on_match policy (block \| redact \| supervise) lint / lint (push) Successful in 1m41s Details test / unit (pull_request) Successful in 30s Details test / integration (pull_request) Successful in 18s Details Give each egress route a policy for what the proxy does when an outbound DLP detector matches a token, defaulting to the supervise flow added in the previous commit. The goal is cutting false-positive friction without weakening default-deny. - redact: scrub the matched value(s) from the body, non-host headers, and path/query via redact_tokens, then re-scan. Forward if clean; fail closed with a 403 if a match remains on a surface redaction can't rewrite (the hostname, or a unicode-evasion token). For routes where a token-shaped value is noise the upstream doesn't need. - block: the original hard 403, never overridable. - supervise (default, unset): hold the request for operator approval. Structural blocks (CRLF, no safelist-able value) stay hard 403s under every policy. Threads outbound_on_match from the bottle manifest (manifest_egress) through the resolved EgressRoute and rendered routes.yaml (egress.py) to the addon's Route (egress_addon_core), and round-trips it via the list-egress-routes introspection endpoint. The allow/egress-block tool descriptions document the new key. Tests: manifest parse/validation, core parse/validation, full manifest->render->addon round-trip for redact. README + PRD 0062 updated. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HnvBjPZC5V7qeQpFbQdDmS	2026-06-24 16:50:13 -04:00
didericis	7f2352287e	PRD 0062: supervisor override for egress token blocks lint / lint (push) Successful in 1m42s Details test / unit (pull_request) Successful in 31s Details test / integration (pull_request) Successful in 16s Details When the outbound DLP catches a token, route the block through the existing supervisor approval queue instead of returning 403 outright. The egress proxy holds the request open until the operator answers, then remembers an approved value for the life of the proxy so the request -- and later ones carrying it -- flow through. Fails closed on rejection, timeout, malformed response, or when supervise is disabled. - ScanResult.matched carries the raw matched substring (sidecar-only; never logged or written to the proposal). scan_outbound and the token detectors take a safe_tokens set and skip approved values, continuing past a safelisted match so a second secret in the same request is still caught. - New egress-token-allow proposal tool, written directly to the queue by the addon (the gitleaks-allow pattern from PRD 0061). build_token_allow _payload renders host/method/path/detector reason + redacted context. - Async request hook polls the queue without stalling the proxy event loop; EGRESS_TOKEN_ALLOW_TIMEOUT_SECONDS (default 300) bounds the wait. - Supervisor TUI renders egress-token-allow like gitleaks-allow: report only, modify unavailable, approval requires a recorded reason. - Unit tests for the matched/safe-tokens plumbing, payload builder, tool constant round-trip, and TUI paths; README + PRD 0062. Closes #261. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01HnvBjPZC5V7qeQpFbQdDmS	2026-06-24 16:12:50 -04:00
didericis	7cb967770e	feat(log): add leveled severity and structured context to log wrappers test / unit (pull_request) Successful in 31s Details test / integration (pull_request) Successful in 16s Details lint / lint (push) Successful in 1m39s Details test / unit (push) Successful in 31s Details test / integration (push) Successful in 16s Details Update Quality Badges / update-badges (push) Successful in 1m32s Details log.py was bare print-to-stderr wrappers with no levels or attributable context (issue #252). Add: - Ordered severities (debug/info/warn/error) gated by BOT_BOTTLE_LOG_LEVEL (default info). debug is silent by default; error always surfaces (nothing sits above it), so the fatal die path stays visible regardless of configured level. - An optional `context` mapping on every wrapper, rendered as a parseable ` [k=v ...]` suffix (keys sorted; whitespace/quoted values quoted) so failures can be filtered and correlated. Default output with no context is byte-identical to the original lines, so the 100+ existing single-string call sites are unaffected. Wires the supervise crash path (the example the issue names) to attach error_type and crash_log context. Adds test_log.py (backward-compat, context rendering, level gating, die surfacing). Closes #252. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcU7nerbg8cVj9R4EkpfLJ	2026-06-24 15:37:57 -04:00
didericis	80eca740d6	docs(research): replace unsourced "20% malicious skills" with cited empirical figures The "~20% of ClawHub skills malicious" claim had no traceable source and is contradicted by the empirical literature. Replace with the Jan 2026 large-scale study (98,380-skill snapshot: 157 confirmed malicious, ~71% credential harvesters, exfiltration overwhelmingly naive) and add the arXiv citation. The corrected figures still support the supply-chain threat point and are defensible under scrutiny. Co-Authored-By: Claude Opus 4.8 <noreply@anthropic.com> Claude-Session: https://claude.ai/code/session_01YcU7nerbg8cVj9R4EkpfLJ	2026-06-24 09:32:19 -04:00