feat(secrets): encrypt egress tokens at rest with per-bottle ENV_VAR_SECRET

Implements the interim secret-provider design (PRD prd-new-secret-provider):
each agent receives a random ENV_VAR_SECRET injected into its container env
at launch. The host uses this key to encrypt each egress auth token value
(HMAC-SHA256 CTR mode, stdlib-only) and store it in a new
bottled_agent_secrets table (one row per env var, key column plaintext for
auditing). The key never touches the DB.

On infra container restart the in-memory token map is lost. launch_consolidated
now calls _reprovision_running_bottles after ensure_running: for each
registered bottle still alive on the gateway network it execs
`printenv ENV_VAR_SECRET` into the agent container and posts the result to the
new POST /bottles/<id>/reprovision_gateway control-plane endpoint, which
decrypts the stored rows and restores _tokens — no manual intervention needed.

Co-Authored-By: Claude Sonnet 4.6 <noreply@anthropic.com>

.coveragerc

@@ -1,6 +1,10 @@
 [run]
 branch = True
 source = .
+# Store paths relative to the project root so .coverage.* files produced on
+# different runners (ubuntu-latest vs self-hosted KVM) can be combined by the
+# coverage job without a [paths] remapping section.
+relative_files = True
 
 [report]
 # Coverage policy: see docs/decisions/0004-coverage-policy.md.

.gitea/workflows/test.yml

@@ -9,10 +9,12 @@
 #   tests/canaries/     — upstream regression canaries; run on a separate
 #                         schedule (see canaries.yml), not here
 #
-# Integration tests run once per backend in separate jobs. Each job sets
-# BOT_BOTTLE_BACKEND explicitly so the test suite uses the right backend.
-# Backends that aren't available on the runner fail the preflight step
-# rather than silently skipping inside the test output.
+# Each test job runs once under coverage and uploads a small .coverage.*
+# artifact. The `coverage` job combines them — no test reruns, no KVM
+# dependency on that job. For main-branch pushes only, the tested rootfs
+# and matching dropbear are uploaded so `publish-infra` can publish the
+# byte-identical artifact that was tested. PRs avoid the ~194 MB rootfs
+# transfer entirely.
 
 name: test
 
@@ -40,53 +42,6 @@ on:
   workflow_dispatch:
 
 jobs:
-  stage-firecracker-inputs:
-    runs-on: [self-hosted, kvm]
-    # Same guard as the other KVM-runner jobs: don't spin the privileged
-    # runner for fork PRs (this only copies a non-secret static binary, but
-    # keep the posture consistent — build-infra/integration/coverage all
-    # depend on it, so gating here gates the whole Firecracker chain).
-    if: >-
-      github.event_name == 'push' ||
-      github.event_name == 'workflow_dispatch' ||
-      (github.event_name == 'pull_request' &&
-       github.event.pull_request.head.repo.full_name == github.repository)
-    steps:
-      - name: Stage the provisioned static dropbear
-        run: |
-          mkdir -p firecracker-inputs
-          cp /var/cache/bot-bottle-fc/dropbear firecracker-inputs/dropbear
-
-      - name: Upload Firecracker build inputs
-        uses: actions/upload-artifact@v3
-        with:
-          name: firecracker-inputs
-          path: firecracker-inputs/
-
-  build-infra:
-    needs: stage-firecracker-inputs
-    runs-on: ubuntu-latest
-    steps:
-      - name: Checkout
-        uses: actions/checkout@v4
-
-      - name: Download Firecracker build inputs
-        uses: actions/download-artifact@v3
-        with:
-          name: firecracker-inputs
-          path: firecracker-inputs
-
-      - name: Build infra candidate from this checkout
-        env:
-          BOT_BOTTLE_FC_DROPBEAR: ${{ github.workspace }}/firecracker-inputs/dropbear
-        run: python3 -m bot_bottle.backend.firecracker.publish_infra --output infra-candidate
-
-      - name: Upload infra candidate
-        uses: actions/upload-artifact@v3
-        with:
-          name: infra-candidate
-          path: infra-candidate/
-
   unit:
     runs-on: ubuntu-latest
     steps:
@@ -101,12 +56,29 @@ jobs:
       - name: Install dev requirements
         run: python3 -m pip install --break-system-packages -r requirements-dev.txt
 
-      - name: Run unit tests
+      - name: Run unit tests with coverage
+        env:
+          COVERAGE_FILE: ${{ github.workspace }}/.coverage.unit
         run: python3 -m coverage run -m unittest discover -t . -s tests/unit -v
 
       - name: Report unit coverage
+        env:
+          COVERAGE_FILE: ${{ github.workspace }}/.coverage.unit
         run: python3 -m coverage report -m
 
+      # upload-artifact@v3's glob skips dotfiles, so a bare `.coverage.unit`
+      # silently uploads nothing ("No files were found"). Stage it under a
+      # non-dot name; the coverage job renames it back before `coverage
+      # combine`. `cp` also fails loudly if coverage never wrote the file.
+      - name: Stage unit coverage for upload
+        run: cp .coverage.unit coverage-unit.dat
+
+      - name: Upload unit coverage artifact
+        uses: actions/upload-artifact@v3
+        with:
+          name: coverage-unit
+          path: coverage-unit.dat
+
   integration-docker:
     runs-on: ubuntu-latest
     steps:
@@ -115,6 +87,9 @@ jobs:
 
       # No actions/setup-python (see the note in the `unit` job); the
       # container's system Python 3.12 runs the stdlib test suite directly.
+      - name: Install coverage
+        run: python3 -m pip install --break-system-packages coverage
+
       - name: Show environment
         run: |
           python3 --version
@@ -124,10 +99,21 @@ jobs:
             echo "docker not on PATH — integration tests will skip"
           fi
 
-      - name: Run integration tests (docker)
+      - name: Run integration tests (docker) with coverage
         env:
           BOT_BOTTLE_BACKEND: docker
-        run: python3 -m unittest discover -t . -s tests/integration -v
+          COVERAGE_FILE: ${{ github.workspace }}/.coverage.docker
+        run: python3 -m coverage run -m unittest discover -t . -s tests/integration -v
+
+      # Non-dot name so upload-artifact's dotfile-skipping glob picks it up.
+      - name: Stage docker coverage for upload
+        run: cp .coverage.docker coverage-docker.dat
+
+      - name: Upload docker coverage artifact
+        uses: actions/upload-artifact@v3
+        with:
+          name: coverage-docker
+          path: coverage-docker.dat
 
   # Integration tests against the Firecracker backend. Runs on a self-hosted
   # KVM runner (label `kvm`) where /dev/kvm and the TAP/nft pool are available.
@@ -137,9 +123,16 @@ jobs:
   #
   # Runner prerequisites (provision once; see README "Firecracker on Linux"):
   # `firecracker` on PATH, `/dev/kvm` accessible, cached kernel +
-  # static dropbear, and the pool as a persistent systemd unit.
+  # static dropbear at /var/cache/bot-bottle-fc/dropbear, and the pool as a
+  # persistent systemd unit.
+  #
+  # The infra candidate is built here directly (no artifact download) to
+  # eliminate the ~70 s ubuntu-latest upload + ~83 s combined download that
+  # the old build-infra → integration-firecracker + coverage chain incurred.
+  # For main-branch pushes the tested rootfs and matching dropbear are
+  # uploaded so publish-infra can publish the byte-identical artifact; PRs
+  # skip those uploads entirely.
   integration-firecracker:
-    needs: build-infra
     runs-on: [self-hosted, kvm]
     if: >-
       github.event_name == 'push' ||
@@ -159,49 +152,65 @@ jobs:
           # range overlap; it prints the exact `backend setup` fix.
           python3 cli.py backend status --backend=firecracker
 
-      - name: Download the candidate built from this checkout
-        uses: actions/download-artifact@v3
-        with:
-          name: infra-candidate
-          path: infra-candidate
+      - name: Build infra candidate from this checkout
+        env:
+          BOT_BOTTLE_FC_DROPBEAR: /var/cache/bot-bottle-fc/dropbear
+        run: python3 -m bot_bottle.backend.firecracker.publish_infra --output infra-candidate --reuse-published
 
       - name: Replace the persistent infra VM with the candidate
         run: python3 -c 'from bot_bottle.backend.firecracker import infra_vm; infra_vm.stop()'
 
-      # No dev-requirements install: the integration suite runs on stdlib
-      # `unittest` (pylint/pyright are lint.yml's concern, not this job's),
-      # and the self-hosted runner's Nix python env has no `pip` module
-      # (`python3 -m pip` → "No module named pip"). Nothing to install.
-      - name: Run integration tests (firecracker)
+      # No dev-requirements install: `coverage` is already provided by the
+      # self-hosted runner's Nix python env, and that env has no `pip`
+      # module to install into anyway.
+      - name: Run integration tests (firecracker) with coverage
         env:
           BOT_BOTTLE_BACKEND: firecracker
           BOT_BOTTLE_INFRA_ARTIFACT_DIR: ${{ github.workspace }}/infra-candidate
-        run: python3 -m unittest discover -t . -s tests/integration -v
+          COVERAGE_FILE: ${{ github.workspace }}/.coverage.firecracker
+        run: python3 -m coverage run -m unittest discover -t . -s tests/integration -v
 
-  # Combined unit+integration coverage + the diff-coverage gate (the hard
-  # gate: new/changed lines >= 90%). See docs/decisions/0004-coverage-policy.md.
+      # Non-dot name so upload-artifact's dotfile-skipping glob picks it up.
+      - name: Stage firecracker coverage for upload
+        run: cp .coverage.firecracker coverage-firecracker.dat
+
+      - name: Upload firecracker coverage artifact
+        uses: actions/upload-artifact@v3
+        with:
+          name: coverage-firecracker
+          path: coverage-firecracker.dat
+
+      # Only upload the large rootfs artifact on main-branch pushes;
+      # PRs avoid the ~194 MB transfer. publish-infra only runs on main
+      # and downloads these to publish the byte-identical tested rootfs.
+      - name: Upload tested rootfs (main branch only)
+        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        uses: actions/upload-artifact@v3
+        with:
+          name: infra-candidate
+          path: infra-candidate/
+
+      - name: Upload dropbear for publish verification (main branch only)
+        if: github.event_name == 'push' && github.ref == 'refs/heads/main'
+        uses: actions/upload-artifact@v3
+        with:
+          name: firecracker-inputs
+          path: /var/cache/bot-bottle-fc/dropbear
+
+  # Combined coverage gate: aggregates .coverage.* artifacts uploaded by each
+  # test job, then runs the diff-coverage gate (new/changed lines >= 90%).
   #
-  # This runs on a self-hosted KVM runner (label `kvm`), NOT ubuntu-latest,
-  # because the Firecracker backend's subprocess/VM orchestration
-  # (launch/boot/SSH/isolation-probe) is covered by the integration suite,
-  # and that suite needs `/dev/kvm` + the provisioned TAP/nft pool — which a
-  # container-based runner doesn't have. On such a runner the firecracker
-  # integration test skips and its ~230 orchestration lines read as
-  # uncovered, so the gate can't pass there.
+  # Runs on ubuntu-latest — no KVM needed, no test reruns. Coverage files use
+  # relative_files = True (.coveragerc) so they combine cleanly across runners.
+  # Each test job sets COVERAGE_FILE to an absolute path so coverage.py writes
+  # to a known location that upload-artifact can find regardless of runner env.
   #
-  # Restricted to the same events as integration-firecracker (same-repo PRs,
-  # push, workflow_dispatch) for the same security reason.
-  #
-  # See #414 for the planned follow-up: artifact-based coverage combination
-  # (run tests once in their respective jobs, combine .coverage files here).
-  #
-  # build-infra creates one candidate from the checkout. This job boots that
-  # same candidate after integration-firecracker has exercised it; the main
-  # push path publishes the identical bytes only after every required job.
+  # Restricted to the same events as integration-firecracker: it depends on
+  # that job's coverage artifact and skips for fork PRs alongside it.
   coverage:
-    needs: [build-infra, integration-firecracker]
+    needs: [unit, integration-docker, integration-firecracker]
     timeout-minutes: 15
-    runs-on: [self-hosted, kvm]
+    runs-on: ubuntu-latest
     if: >-
       github.event_name == 'push' ||
       github.event_name == 'workflow_dispatch' ||
@@ -213,29 +222,37 @@ jobs:
         with:
           fetch-depth: 0
 
-      - name: Preflight — Firecracker host is ready
-        run: |
-          command -v firecracker >/dev/null || {
-            echo "firecracker not on PATH — provision the runner (README: Firecracker on Linux)"; exit 1; }
-          test -e /dev/kvm || { echo "/dev/kvm missing — KVM not available on this runner"; exit 1; }
-          # `backend status` exits non-zero unless the TAP pool is up + no
-          # range overlap; it prints the exact `backend setup` fix.
-          python3 cli.py backend status --backend=firecracker
+      - name: Install coverage
+        run: python3 -m pip install --break-system-packages coverage
 
-      - name: Download the candidate already exercised by integration
+      - name: Download unit coverage artifact
         uses: actions/download-artifact@v3
         with:
-          name: infra-candidate
-          path: infra-candidate
+          name: coverage-unit
+          path: ${{ github.workspace }}
+
+      - name: Download docker coverage artifact
+        uses: actions/download-artifact@v3
+        with:
+          name: coverage-docker
+          path: ${{ github.workspace }}
+
+      - name: Download firecracker coverage artifact
+        uses: actions/download-artifact@v3
+        with:
+          name: coverage-firecracker
+          path: ${{ github.workspace }}
+
+      # Rename the non-dot upload names back to the .coverage.* files that
+      # `coverage combine` discovers (see the staging steps in each test job).
+      - name: Reassemble coverage data files
+        run: |
+          mv coverage-unit.dat .coverage.unit
+          mv coverage-docker.dat .coverage.docker
+          mv coverage-firecracker.dat .coverage.firecracker
 
-      # No dev-requirements install: `coverage` is already provided by the
-      # self-hosted runner's Nix python env, and that env has no `pip`
-      # module to install into anyway. `scripts/coverage.sh` +
-      # `diff_coverage.py` need only `coverage` (not pylint/pyright).
       - name: Combined coverage (unit + integration, incl. firecracker)
-        env:
-          BOT_BOTTLE_CI_INFRA_ARTIFACT_DIR: ${{ github.workspace }}/infra-candidate
-        run: PYTHON=python3 bash scripts/coverage.sh critical
+        run: PYTHON=python3 bash scripts/coverage.sh aggregate critical
 
       - name: Diff-coverage gate (changed lines >= 90%)
         run: |
@@ -243,14 +260,14 @@ jobs:
           python3 scripts/diff_coverage.py --base origin/main --min 90
 
   publish-infra:
-    needs: [stage-firecracker-inputs, build-infra, unit, integration-docker, integration-firecracker, coverage]
+    needs: [unit, integration-docker, integration-firecracker, coverage]
     runs-on: ubuntu-latest
     if: github.event_name == 'push' && github.ref == 'refs/heads/main'
     steps:
       - name: Checkout the tested revision
         uses: actions/checkout@v4
 
-      - name: Download the tested candidate
+      - name: Download the tested rootfs
         uses: actions/download-artifact@v3
         with:
           name: infra-candidate
@@ -258,9 +275,10 @@ jobs:
 
       # publish_infra re-derives the version from the checkout to confirm the
       # bundle matches before uploading, and the version hashes the dropbear
-      # bytes. Stage the SAME dropbear build-infra used, or the recheck
-      # computes a "<missing>"-dropbear version and rejects the candidate.
-      - name: Download the staged dropbear (matches build-infra's version)
+      # bytes. Download the SAME dropbear integration-firecracker used, or
+      # the recheck computes a "<missing>"-dropbear version and rejects the
+      # candidate.
+      - name: Download the staged dropbear (matches build's version)
         uses: actions/download-artifact@v3
         with:
           name: firecracker-inputs

.gitea/workflows/update-badges.yml

@@ -14,6 +14,9 @@ on:
 jobs:
   update-badges:
     runs-on: ubuntu-latest
+    permissions:
+      contents: write
+
     steps:
       - uses: actions/checkout@v3
         with:

Dockerfile.infra

@@ -1,45 +1,22 @@
-# Firecracker single infra-VM image (PRD 0070 Stage B).
+# Shared infra image: gateway data plane + orchestrator control plane.
 #
-# The per-host infra VM runs the orchestrator control plane, the gateway
-# data plane, AND builds agent images (buildah) — all in one microVM (see
-# backend/firecracker/infra_vm.py). It composes:
-#   * FROM the gateway image (mitmproxy / git / gitleaks / supervise + the
-#     flat daemon modules) — now trixie-based, so buildah 1.39 is available;
-#   * `COPY --from` the orchestrator image's content (the single definition
-#     of the control-plane payload — see Dockerfile.orchestrator), so this
-#     VM and the docker backend share one orchestrator definition; and
-#   * buildah, installed HERE only (the docker orchestrator/gateway images
-#     never carry it).
+# Used directly by the Docker backend (run as one `bot-bottle-infra`
+# container, replacing the prior two-container split). The Firecracker
+# backend extends this via Dockerfile.infra.fc, adding buildah/crun/
+# netavark for in-VM agent-image building.
+#
+# Dockerfile.orchestrator is the single definition of the orchestrator
+# content (the lean `bot_bottle` package on python:3.12-slim). Both this
+# image and Dockerfile.infra.fc pull it in via `COPY --from`.
 #
 # multi-`FROM` can't union two bases (that's multi-stage, not multiple
 # inheritance), so the orchestrator content is pulled in via `COPY --from`
 # rather than a second base. Both images share the trixie `python:3.12-slim`
 # base, so the copy is clean (same python; future installed deps copy too).
-#
-# The docker backend keeps orchestrator + gateway as separate images; this
-# combined image exists only for the Firecracker single-VM cut. Splitting a
-# service back into its own VM later is a routing change, not a repackaging
-# (PRD 0070's "secret concentration"; a disposable builder can boot from
-# this same image on its own TAP).
 FROM bot-bottle-gateway:latest
 
-# --- in-VM agent-image builder (PRD 0069 Stage 3) -------------------
-# The Firecracker backend builds users' agent Dockerfiles *inside this VM*
-# with buildah (rootless, daemonless) instead of on the host — no host
-# Docker daemon, no root-equivalent `docker` group. `crun` is the OCI
-# runtime; `netavark` + `aardvark-dns` are the network backend for `FROM`
-# pulls + `RUN` egress. Requires the trixie base (buildah 1.39: bookworm's
-# 1.28 can't parse Dockerfile heredocs that agent images use).
-RUN apt-get update \
- && apt-get install -y --no-install-recommends \
-      buildah crun netavark aardvark-dns \
- && rm -rf /var/lib/apt/lists/*
-# vfs + chroot: buildah works as root in the bare microVM (no
-# fuse-overlayfs / overlay module / subuid maps). Matches image_builder.
-ENV STORAGE_DRIVER=vfs \
-    BUILDAH_ISOLATION=chroot
-
-# The orchestrator content, pulled from its single definition. The gateway
-# image already has the flat daemon modules under /app; this adds the full
-# `bot_bottle` package so `python3 -m bot_bottle.orchestrator` resolves.
+# The orchestrator content, from its single definition. The gateway image
+# already has the flat daemon modules under /app; this adds the full
+# `bot_bottle` package so `python3 -m bot_bottle.orchestrator` resolves —
+# used by gateway_init when BOT_BOTTLE_GATEWAY_DAEMONS includes `orchestrator`.
 COPY --from=bot-bottle-orchestrator:latest /app/bot_bottle /app/bot_bottle

Dockerfile.infra.fc

@@ -0,0 +1,23 @@
+# Firecracker infra VM image (PRD 0070 Stage B).
+#
+# Extends the shared infra base (Dockerfile.infra: gateway + orchestrator
+# control plane) with the in-VM agent-image builder. The Firecracker backend
+# builds users' agent Dockerfiles *inside this VM* with buildah (rootless,
+# daemonless) instead of on the host — no host Docker daemon, no
+# root-equivalent `docker` group.
+#
+# Requires the trixie base from bot-bottle-gateway (buildah 1.39: bookworm's
+# 1.28 can't parse Dockerfile heredocs that agent images use).
+#
+# `crun` is the OCI runtime; `netavark` + `aardvark-dns` are the network
+# backend for `FROM` pulls + `RUN` egress. `vfs` + `chroot`: buildah works
+# as root in the bare microVM (no fuse-overlayfs / overlay module / subuid
+# maps). Matches image_builder.
+FROM bot-bottle-infra:latest
+
+RUN apt-get update \
+ && apt-get install -y --no-install-recommends \
+      buildah crun netavark aardvark-dns \
+ && rm -rf /var/lib/apt/lists/*
+ENV STORAGE_DRIVER=vfs \
+    BUILDAH_ISOLATION=chroot

README.md

@@ -75,22 +75,6 @@ On compatible macOS hosts, the default backend requires Apple's `container` CLI
 
 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.
 
-> **Experimental containers-in-bottle spike (#392):** a bottle may set
-> `docker_access: true`. On the macOS backend this starts a guest-local,
-> rootless **podman** service after the bottle is registered, exposing its
-> Docker-compatible API socket — the agent still uses `docker` and `docker
-> compose`. It does not mount Docker Desktop's socket or add outer VM
-> capabilities. Rootless Docker was tried first and does not work here at
-> all: Apple Container's capability bounding set omits `CAP_SYS_ADMIN`,
-> which the kernel requires to write a multi-range `uid_map`. See
-> [`docs/research/rootless-docker-in-apple-container-spike.md`](docs/research/rootless-docker-in-apple-container-spike.md).
->
-> The tradeoff to understand before enabling it: podman avoids that
-> requirement by falling back to a single-UID mapping, so nested containers
-> provide **no isolation from the agent itself** — `root` inside a nested
-> container is the agent user outside it. Nested containers are a build/test
-> convenience, not a security boundary. The bottle remains the boundary.
-
 ### Firecracker on Linux
 
 On Linux, a KVM-capable host defaults to the Firecracker backend. It needs:

bot_bottle/agent_provider.py

@@ -45,6 +45,10 @@ PROVIDER_TEMPLATES = frozenset({PROVIDER_CLAUDE, PROVIDER_CODEX, PROVIDER_PI})
 # forward_host_credentials is enabled. Pipelock must pass these through
 # (no TLS MITM) or its header DLP blocks the injected JWT.
 CODEX_HOST_CREDENTIAL_HOSTS = ("api.openai.com", "chatgpt.com")
+
+# Host that egress injects the host Claude bearer on when Claude
+# forward_host_credentials is enabled.
+CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)
 PromptMode = Literal[
     "append_file",
     "read_prompt_file",
@@ -257,7 +261,28 @@ class AgentProvider(ABC):
         Default: Debian/node — writes the git-gate insteadOf gitconfig
         and sets user.name/email as node. Workspace copy runs through
         BottleBackend.provision_workspace against the running bottle."""
-        from .log import info
+        from .log import die, info
+
+        # Firecracker exports image rootfs files through an unprivileged host
+        # tar extraction, so image-time ownership of XDG directories is not
+        # preserved.  Git consults ~/.config/git even when the actual config
+        # is ~/.gitconfig; an unreadable directory there can prevent the
+        # git-gate insteadOf rules below from taking effect.  Repair this at
+        # runtime, after every backend's copy/export path has completed.
+        git_xdg_dir = f"{plan.guest_home}/.config/git"
+        repair = bottle.exec(
+            f"chown node:node {shlex.quote(plan.guest_home)} && "
+            f"chmod 755 {shlex.quote(plan.guest_home)} && "
+            f"mkdir -p {shlex.quote(git_xdg_dir)} && "
+            f"chown -R node:node {shlex.quote(f'{plan.guest_home}/.config')} && "
+            f"chmod -R u+rwX,go+rX {shlex.quote(f'{plan.guest_home}/.config')}",
+            user="root",
+        )
+        if repair.returncode != 0:
+            die(
+                "git provisioning: could not make the runtime Git config "
+                f"directory readable: {(repair.stderr or repair.stdout).strip()}"
+            )
 
         manifest_bottle = plan.manifest.bottle
         if manifest_bottle.git:
@@ -280,11 +305,27 @@ class AgentProvider(ABC):
                 f"{len(manifest_bottle.git)} insteadOf rule(s)"
             )
             bottle.cp_in(str(config_file), guest_gitconfig)
-            bottle.exec(
+            permissions = bottle.exec(
                 f"chown node:node {shlex.quote(guest_gitconfig)} && "
                 f"chmod 644 {shlex.quote(guest_gitconfig)}",
                 user="root",
             )
+            if permissions.returncode != 0:
+                die(
+                    "git provisioning: could not set ownership on "
+                    f"{guest_gitconfig}: "
+                    f"{(permissions.stderr or permissions.stdout).strip()}"
+                )
+            configured = bottle.exec(
+                "git config --global --get-regexp '^url\\..*\\.insteadof$'",
+                user="node",
+            )
+            if configured.returncode != 0:
+                die(
+                    "git provisioning: the runtime user cannot read the "
+                    f"git-gate insteadOf rules from {guest_gitconfig}: "
+                    f"{(configured.stderr or configured.stdout).strip()}"
+                )
 
         gu = manifest_bottle.git_user
         if not gu.is_empty():

bot_bottle/backend/__init__.py

@@ -37,10 +37,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager
+from contextlib import AbstractContextManager, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import TYPE_CHECKING, Any, Generic, Sequence, TypeVar
+from typing import TYPE_CHECKING, Any, Generator, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -83,6 +83,9 @@ class BottleSpec:
     # True when launched via --headless (no TTY, no interactive prompts).
     # The git-gate host-key preflight uses this to error rather than prompt.
     headless: bool = False
+    # Image startup policy. "fresh" preserves the normal build path;
+    # "cached" reuses the current local image/artifact without rebuilding.
+    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -278,6 +281,18 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
 CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan)
 
 
+@dataclass(frozen=True)
+class BottleImages:
+    """Resolved image references (or artifact paths) for a bottle launch.
+
+    For Docker/macOS-container backends, `agent` and `sidecar` are string
+    image refs. For the smolmachines backend they are Path objects pointing
+    to pre-built `.smolmachine` artifacts."""
+
+    agent: str | Path
+    sidecar: str | Path = ""
+
+
 class BottleBackend(ABC, Generic[PlanT, CleanupT]):
     """Abstract base for selectable bottle backends. Concrete subclasses
     (e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -437,9 +452,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
+    def prelaunch_checks(self, plan: PlanT) -> None:
+        """Raise StaleImageError if any cached image used by this plan is stale.
+        No-op default; backends override to call the shared check_stale*
+        helpers on their image/artifact timestamps. Called by the CLI before
+        launch so the operator can be prompted outside the launch context."""
+
+    @contextmanager
+    def launch(self, plan: PlanT) -> Generator[Bottle, None, None]:
+        """Template: build or load images, then delegate to _launch_impl."""
+        images = self._build_or_load_images(plan)
+        with self._launch_impl(plan, images) as bottle:
+            yield bottle
+
     @abstractmethod
-    def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
-        """Build/run the bottle and yield a handle; tear down on exit."""
+    def _build_or_load_images(self, plan: PlanT) -> BottleImages:
+        """Return the agent and sidecar image references (or artifact paths)
+        for this plan, building fresh images when the policy requires it."""
+
+    @abstractmethod
+    def _launch_impl(self, plan: PlanT, images: BottleImages) -> AbstractContextManager[Bottle]:
+        """Bring up the bottle using pre-resolved images; yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
         """Copy host-side files (CA cert, prompt, skills, .git) into

bot_bottle/backend/consolidated_util.py

@@ -0,0 +1,68 @@
+"""Shared helpers for the consolidated launch sequence (PRD 0070).
+
+Logic that was duplicated across the docker, macos_container, and
+firecracker consolidated_launch modules — extracted so each backend
+imports it rather than re-implementing it.
+"""
+
+from __future__ import annotations
+
+import dataclasses
+
+from ..egress import EgressPlan
+from ..git_gate import GitGatePlan
+from ..orchestrator.client import OrchestratorClient, RegisteredBottle
+from ..orchestrator.registration import registration_inputs
+from ..orchestrator.secret_store import new_env_var_secret
+from .docker.gateway_provision import GatewayTransport, deprovision_git_gate, provision_git_gate
+
+
+def provision_bottle(
+    client: OrchestratorClient,
+    source_ip: str,
+    egress_plan: EgressPlan,
+    git_gate_plan: GitGatePlan,
+    transport: GatewayTransport,
+    *,
+    image_ref: str = "",
+    tokens: dict[str, str] | None = None,
+) -> RegisteredBottle:
+    """Register the bottle and provision its git-gate state. Rolls back the
+    registration if provisioning fails so no orphan is left.
+
+    Generates a fresh ENV_VAR_SECRET, passes it to the orchestrator so it can
+    encrypt the token values at rest, and stamps the secret onto the returned
+    ``RegisteredBottle`` so callers can inject it into the agent container's
+    environment."""
+    inputs = registration_inputs(egress_plan)
+    env_var_secret = new_env_var_secret()
+    reg = client.register_bottle(
+        source_ip, image_ref=image_ref, policy=inputs.policy,
+        metadata=inputs.metadata, tokens=tokens, env_var_secret=env_var_secret,
+    )
+    try:
+        provision_git_gate(transport, reg.bottle_id, git_gate_plan)
+    except Exception:
+        client.teardown_bottle(reg.bottle_id)
+        raise
+    return dataclasses.replace(reg, env_var_secret=env_var_secret)
+
+
+def teardown_consolidated(
+    bottle_id: str,
+    transport: GatewayTransport,
+    *,
+    orchestrator_url: str,
+    timeout: float | None = None,
+) -> None:
+    """Deregister the bottle and remove its git-gate state. Both steps are
+    idempotent so this is safe from a cleanup trap."""
+    from ..orchestrator.config_store import DEFAULT_TEARDOWN_TIMEOUT_SECONDS
+    OrchestratorClient(
+        orchestrator_url,
+        timeout=timeout if timeout is not None else DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
+    ).teardown_bottle(bottle_id)
+    deprovision_git_gate(transport, bottle_id)
+
+
+__all__ = ["provision_bottle", "teardown_consolidated"]

bot_bottle/backend/docker/backend.py

@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -100,9 +100,15 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
+    def prelaunch_checks(self, plan: DockerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
+    def _build_or_load_images(self, plan: DockerBottlePlan) -> BottleImages:
+        return _launch.build_or_load_images(plan)
+
     @contextmanager
-    def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+    def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def ensure_orchestrator(self) -> str:

bot_bottle/backend/docker/bottle_plan.py

@@ -39,6 +39,10 @@ class DockerBottlePlan(BottlePlan):
     # (egress proxy credentials, git-gate/supervise headers); set by launch
     # from the orchestrator registration. Empty pre-registration.
     identity_token: str = ""
+    # Encryption key for the agent's stored egress secrets; injected into the
+    # agent container as ENV_VAR_SECRET via the compose subprocess env (bare
+    # name — value never written to the compose file). Empty pre-registration.
+    env_var_secret: str = ""
 
     @property
     def container_name(self) -> str:

bot_bottle/backend/docker/consolidated_compose.py

@@ -17,6 +17,7 @@ from __future__ import annotations
 from typing import Any
 
 from ...egress import egress_agent_env_entries
+from ...orchestrator.secret_store import ENV_VAR_SECRET_NAME
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
 from .bottle_plan import DockerBottlePlan
 from .egress import EGRESS_PORT
@@ -58,6 +59,10 @@ def consolidated_agent_compose(
     # the secret value never lands on argv or in the compose file.
     for name in sorted(plan.forwarded_env.keys()):
         env.append(name)
+    # ENV_VAR_SECRET: bare name so the value comes from the compose subprocess
+    # env (set in launch.py) and is never written to the compose file on disk.
+    if getattr(plan, "env_var_secret", ""):
+        env.append(ENV_VAR_SECRET_NAME)
     env.extend(egress_agent_env_entries(plan.egress_plan))
 
     service: dict[str, Any] = {

bot_bottle/backend/docker/consolidated_launch.py

@@ -1,19 +1,13 @@
 """Consolidated bottle launch sequence for the docker backend (PRD 0070).
 
-Composes the orchestrator primitives into the register/teardown sequence that
-replaces the per-bottle gateway:
+Composes the orchestrator primitives into the register/teardown sequence:
 
-  1. ensure the orchestrator control plane + shared gateway are up;
-  2. allocate the bottle a pinned source IP on the gateway network (the
-     attribution key), skipping the gateway's own address + live bottles;
-  3. register it (egress policy blob + slug metadata) → bottle id + identity
-     token;
-  4. provision its git-gate repos/creds into the running gateway.
+  1. ensure the single infra container (control plane + gateway) is up;
+  2. allocate the bottle a pinned source IP on the gateway network;
+  3. register it and provision its git-gate repos/creds into the gateway.
 
-It returns a `LaunchContext` with everything the agent container needs to
-attach — network, pinned IP, the gateway's address (its proxy target), the
-orchestrator URL, and the identity token. The agent `docker run` itself is
-the backend's job (it owns provider provisioning); this owns the
+Returns a `LaunchContext` with everything the agent container needs to
+attach. The agent `docker run` itself is the backend's job; this owns the
 orchestrator-facing wiring so that sequence stays testable in isolation.
 """
 
@@ -21,19 +15,18 @@ from __future__ import annotations
 
 from dataclasses import dataclass
 
+from ... import log
 from ...docker_cmd import run_docker
 from ...egress import EgressPlan
 from ...git_gate import GitGatePlan
 from ...orchestrator.client import OrchestratorClient
-from ...orchestrator.gateway import GATEWAY_NAME, GATEWAY_NETWORK
-from ...orchestrator.lifecycle import OrchestratorService
-from ...orchestrator.registration import registration_inputs
+from ...orchestrator.gateway import GATEWAY_NETWORK
+from ...orchestrator.lifecycle import INFRA_NAME, OrchestratorService
+from ...orchestrator.secret_store import ENV_VAR_SECRET_NAME
+from ..consolidated_util import provision_bottle
+from ..consolidated_util import teardown_consolidated as _teardown_util
+from .gateway_provision import DockerGatewayTransport
 from .gateway_net import next_free_ip
-from .gateway_provision import (
-    DockerGatewayTransport,
-    deprovision_git_gate,
-    provision_git_gate,
-)
 
 
 class ConsolidatedLaunchError(RuntimeError):
@@ -50,6 +43,7 @@ class LaunchContext:
     network: str            # the shared gateway network to attach to
     gateway_ip: str         # the gateway's address — the agent's proxy target
     orchestrator_url: str
+    env_var_secret: str = ""  # encryption key injected into the agent's env
 
 
 def _network_cidr(network: str) -> str:
@@ -75,28 +69,85 @@ def _container_ip(name: str, network: str) -> str:
     ip = proc.stdout.strip()
     if proc.returncode != 0 or not ip:
         raise ConsolidatedLaunchError(
-            f"gateway {name} has no address on {network}: {proc.stderr.strip()}"
+            f"container {name} has no address on {network}: {proc.stderr.strip()}"
         )
     return ip
 
 
 def _network_container_ips(network: str) -> list[str]:
     """Every address currently assigned on the gateway network — the ground
-    truth for "in use": the gateway + orchestrator infrastructure containers
-    and every live agent. Read from the network so a new bottle can't collide
-    with anything actually attached (a registry-only view would miss the
-    orchestrator/gateway containers)."""
+    truth for "in use": the infra container and every live agent. Read from
+    the network so a new bottle can't collide with anything actually attached."""
     proc = run_docker([
         "docker", "network", "inspect", "--format",
         "{{range .Containers}}{{.IPv4Address}} {{end}}", network,
     ])
     ips: list[str] = []
     for entry in proc.stdout.split():
-        # entries look like "172.20.0.2/16" — keep the address.
         ips.append(entry.split("/", 1)[0])
     return ips
 
 
+def _reprovision_running_bottles(
+    orchestrator_url: str,
+    network: str = GATEWAY_NETWORK,
+    infra_name: str = INFRA_NAME,
+) -> None:
+    """Re-inject egress tokens for any registered bottles that lost their
+    in-memory tokens (e.g., after an infra container restart).
+
+    For each registered bottle whose source IP maps to a live container on the
+    gateway network, reads ENV_VAR_SECRET via ``docker exec … printenv`` and
+    calls ``POST /bottles/<id>/reprovision_gateway``. Idempotent — a no-op
+    when the orchestrator already has all tokens loaded. Best-effort: a single
+    container exec failure never blocks a new bottle launch."""
+    client = OrchestratorClient(orchestrator_url)
+    bottles = client.list_bottles()
+    if not bottles:
+        return
+
+    # Build {source_ip: container_name} from live containers on the gateway
+    # network, excluding the infra container itself.
+    proc = run_docker([
+        "docker", "network", "inspect",
+        "--format", "{{range .Containers}}{{.Name}} {{.IPv4Address}}\n{{end}}",
+        network,
+    ])
+    ip_to_container: dict[str, str] = {}
+    for line in proc.stdout.splitlines():
+        parts = line.strip().split()
+        if len(parts) >= 2 and parts[0] != infra_name:
+            ip = parts[1].split("/", 1)[0]
+            if ip:
+                ip_to_container[ip] = parts[0]
+
+    reprovisioned = 0
+    for bottle in bottles:
+        bottle_id = bottle.get("bottle_id")
+        source_ip = bottle.get("source_ip")
+        if not isinstance(bottle_id, str) or not isinstance(source_ip, str):
+            continue
+        container_name = ip_to_container.get(source_ip)
+        if not container_name:
+            continue
+        proc = run_docker(
+            ["docker", "exec", container_name, "printenv", ENV_VAR_SECRET_NAME]
+        )
+        if proc.returncode != 0 or not proc.stdout.strip():
+            continue
+        try:
+            if client.reprovision_gateway(bottle_id, proc.stdout.strip()):
+                reprovisioned += 1
+        except Exception:  # noqa: BLE001 — best-effort, never block a launch
+            pass
+
+    if reprovisioned:
+        log.info(
+            "reprovisioned egress tokens",
+            context={"count": reprovisioned},
+        )
+
+
 def launch_consolidated(
     egress_plan: EgressPlan,
     git_gate_plan: GitGatePlan,
@@ -104,33 +155,29 @@ def launch_consolidated(
     image_ref: str = "",
     tokens: dict[str, str] | None = None,
     service: OrchestratorService | None = None,
-    gateway_name: str = GATEWAY_NAME,
+    infra_name: str = INFRA_NAME,
     network: str = GATEWAY_NETWORK,
 ) -> LaunchContext:
-    """Ensure the orchestrator + gateway are up, allocate + register the
-    bottle, and provision its git-gate state. Returns the agent's attach
-    context. Raises `ConsolidatedLaunchError` (or the primitives' own errors)
-    if any step fails — the caller tears down on failure."""
+    """Ensure the infra container is up, allocate + register the bottle, and
+    provision its git-gate state. Returns the agent's attach context.
+
+    Also reprovisiones egress tokens for any already-running bottles that lost
+    their in-memory credentials (e.g. after an infra container restart), so
+    they regain egress access before the new bottle is registered."""
     service = service or OrchestratorService()
     url = service.ensure_running()
+    _reprovision_running_bottles(url, network=network, infra_name=infra_name)
     client = OrchestratorClient(url)
 
     cidr = _network_cidr(network)
-    gateway_ip = _container_ip(gateway_name, network)
+    gateway_ip = _container_ip(infra_name, network)
     source_ip = next_free_ip(cidr, _network_container_ips(network))
 
-    inputs = registration_inputs(egress_plan)
-    reg = client.register_bottle(
-        source_ip, image_ref=image_ref, policy=inputs.policy,
-        metadata=inputs.metadata, tokens=tokens,
+    transport = DockerGatewayTransport(infra_name)
+    reg = provision_bottle(
+        client, source_ip, egress_plan, git_gate_plan, transport,
+        image_ref=image_ref, tokens=tokens,
     )
-    try:
-        provision_git_gate(
-            DockerGatewayTransport(gateway_name), reg.bottle_id, git_gate_plan)
-    except Exception:
-        # Roll the registration back so a provisioning failure leaves no orphan.
-        client.teardown_bottle(reg.bottle_id)
-        raise
     return LaunchContext(
         bottle_id=reg.bottle_id,
         identity_token=reg.identity_token,
@@ -138,24 +185,17 @@ def launch_consolidated(
         network=network,
         gateway_ip=gateway_ip,
         orchestrator_url=url,
+        env_var_secret=reg.env_var_secret,
     )
 
 
 def teardown_consolidated(
-    bottle_id: str,
-    *,
-    orchestrator_url: str,
-    gateway_name: str = GATEWAY_NAME,
+    bottle_id: str, *, orchestrator_url: str, infra_name: str = INFRA_NAME,
     timeout: float | None = None,
 ) -> None:
-    """Deregister the bottle and remove its git-gate state from the gateway.
-    Both steps are idempotent so this is safe from a cleanup trap."""
-    from ...orchestrator.config_store import DEFAULT_TEARDOWN_TIMEOUT_SECONDS
-    OrchestratorClient(
-        orchestrator_url,
-        timeout=timeout if timeout is not None else DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
-    ).teardown_bottle(bottle_id)
-    deprovision_git_gate(DockerGatewayTransport(gateway_name), bottle_id)
+    """Deregister the bottle and remove its git-gate state. Idempotent."""
+    _teardown_util(bottle_id, DockerGatewayTransport(infra_name),
+                   orchestrator_url=orchestrator_url, timeout=timeout)
 
 
 __all__ = [

bot_bottle/backend/docker/launch.py

@@ -42,7 +42,9 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale
+from ...log import die, info, warn
+from .. import BottleImages
 from . import util as docker_mod
 from .bottle import DockerBottle
 from .bottle_plan import DockerBottlePlan
@@ -71,16 +73,47 @@ from ...orchestrator.gateway import DockerGateway
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 
 
+def build_or_load_images(plan: DockerBottlePlan) -> BottleImages:
+    """Resolve the agent image ref for this plan.
+
+    Returns the committed snapshot if one exists, the cached image when the
+    policy is 'cached', or builds a fresh image and returns that."""
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return BottleImages(agent=committed)
+    if plan.spec.image_policy == "cached":
+        if not docker_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        return BottleImages(agent=plan.image)
+    docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    docker_mod.verify_agent_image(
+        plan.image, runtime_for(plan.agent_provider_template).smoke_test,
+    )
+    return BottleImages(agent=plan.image)
+
+
 @contextmanager
 def launch(
     plan: DockerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
 ) -> Generator[DockerBottle, None, None]:
-    """Build, launch, and provision a Docker bottle via compose.
-    Teardown on exit."""
+    """Launch and provision a Docker bottle via compose. Teardown on exit."""
     stack = ExitStack()
 
+    # Stamp the resolved agent image ref into the plan so compose rendering
+    # picks up the right image (may be a committed snapshot or cached ref).
+    plan = dataclasses.replace(
+        plan,
+        agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
+    )
+
     _bottle_for_revoke = plan.manifest.bottle
     _git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
 
@@ -97,25 +130,6 @@ def launch(
         )
 
     try:
-        # Step 1: agent image. Use a committed snapshot when one exists
-        # and is present in the local daemon; otherwise build from the
-        # Dockerfile. (The gateway image is built by the orchestrator.)
-        committed = read_committed_image(plan.slug)
-        if committed and docker_mod.image_exists(committed):
-            info(f"using committed image {committed!r}")
-            plan = dataclasses.replace(
-                plan,
-                agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
-            )
-        else:
-            docker_mod.build_image(
-                plan.image, _REPO_DIR,
-                dockerfile=plan.dockerfile_path,
-            )
-            docker_mod.verify_agent_image(
-                plan.image, runtime_for(plan.agent_provider_template).smoke_test,
-            )
-
         # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any, before
         # provisioning the bottle's repos into the shared gateway.
         git_gate_plan = plan.git_gate_plan
@@ -172,6 +186,7 @@ def launch(
             agent_git_gate_url=git_gate_url,
             agent_supervise_url=supervise_url,
             identity_token=ctx.identity_token,
+            env_var_secret=ctx.env_var_secret,
         )
 
         # Step 5: render + up the agent-only compose, pinned on the shared
@@ -184,7 +199,12 @@ def launch(
         project = compose_project_name(plan.slug)
         # Forwarded vars (OAuth token, host interpolations) flow through the
         # subprocess env as bare names so values never land in the file.
+        # ENV_VAR_SECRET follows the same pattern: bare name in the compose
+        # spec, value only in the subprocess env so it is never written to disk.
         compose_env: dict[str, str] = {**os.environ, **plan.forwarded_env}
+        if plan.env_var_secret:
+            from ...orchestrator.secret_store import ENV_VAR_SECRET_NAME
+            compose_env[ENV_VAR_SECRET_NAME] = plan.env_var_secret
         info(
             f"docker compose up -d  (project {project}, agent on shared "
             f"gateway {ctx.gateway_ip}, ip {ctx.source_ip})"
@@ -211,3 +231,21 @@ def launch(
         yield bottle
     finally:
         teardown()
+
+
+def stale_checks(plan: DockerBottlePlan) -> None:
+    """Raise StaleImageError if a cached image is older than the configured
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        ts = docker_mod.image_created_at(committed)
+        if ts is not None:
+            check_stale(f"agent image {committed!r}", ts)
+        return
+    if docker_mod.image_exists(plan.image):
+        ts = docker_mod.image_created_at(plan.image)
+        if ts is not None:
+            check_stale(f"agent image {plan.image!r}", ts)

bot_bottle/backend/docker/util.py

@@ -5,6 +5,7 @@ existence, and building images."""
 from __future__ import annotations
 
 import os
+from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -197,3 +198,46 @@ def commit_container(container_name: str, image_tag: str) -> None:
             f"{(result.stderr or '').strip() or '<no stderr>'}"
         )
     info(f"committed {container_name!r} → {image_tag!r}")
+
+
+def image_created_at(ref: str) -> datetime | None:
+    """Return Docker's image Created timestamp as an aware UTC datetime, or
+    None when the field is absent or unparseable. Callers should skip the
+    stale check when None is returned."""
+    r = subprocess.run(
+        ["docker", "image", "inspect", "--format", "{{.Created}}", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if r.returncode != 0:
+        die(
+            f"docker image inspect for {ref!r} failed: "
+            f"{(r.stderr or '').strip() or '<no stderr>'}"
+        )
+    raw = r.stdout.strip()
+    if not raw:
+        return None
+    try:
+        return _parse_docker_timestamp(raw)
+    except ValueError:
+        return None
+
+
+def _parse_docker_timestamp(raw: str) -> datetime:
+    text = raw.strip()
+    if text.endswith("Z"):
+        text = text[:-1] + "+00:00"
+    dot = text.find(".")
+    if dot != -1:
+        tz_plus = text.find("+", dot)
+        tz_minus = text.find("-", dot)
+        tz_candidates = [pos for pos in (tz_plus, tz_minus) if pos != -1]
+        if tz_candidates:
+            tz_pos = min(tz_candidates)
+            frac = text[dot + 1:tz_pos]
+            text = text[:dot + 1] + frac[:6].ljust(6, "0") + text[tz_pos:]
+    dt = datetime.fromisoformat(text)
+    if dt.tzinfo is None:
+        dt = dt.replace(tzinfo=timezone.utc)
+    return dt.astimezone(timezone.utc)

bot_bottle/backend/firecracker/backend.py

@@ -18,7 +18,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...manifest import Manifest
 from ...supervise import SupervisePlan
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -92,11 +92,18 @@ class FirecrackerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _build_or_load_images(self, plan: FirecrackerBottlePlan) -> BottleImages:
+        return BottleImages(agent=_launch.build_or_load_agent_base(plan))
+
+    def prelaunch_checks(self, plan: FirecrackerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
-        self, plan: FirecrackerBottlePlan
+    def _launch_impl(
+        self, plan: FirecrackerBottlePlan, images: BottleImages,
     ) -> Generator[FirecrackerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        assert isinstance(images.agent, Path)
+        with _launch.launch(plan, images.agent, provision=self.provision) as bottle:
             yield bottle
 
     def prepare_cleanup(self) -> FirecrackerBottleCleanupPlan:

bot_bottle/backend/firecracker/consolidated_launch.py

@@ -33,8 +33,7 @@ from ...orchestrator.client import OrchestratorClient
 from ...orchestrator.lifecycle import (
     OrchestratorStartError,  # re-exported so callers can catch it
 )
-from ...orchestrator.registration import registration_inputs
-from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
+from ..consolidated_util import provision_bottle, teardown_consolidated as _teardown_util
 from . import infra_vm
 
 
@@ -51,6 +50,7 @@ class LaunchContext:
     source_ip: str          # the VM's guest IP — the attribution key
     gateway_ca_pem: str     # the shared gateway CA the provisioner installs
     orchestrator_url: str
+    env_var_secret: str = ""  # encryption key injected into the agent's env
 
 
 def launch_consolidated(
@@ -68,18 +68,11 @@ def launch_consolidated(
     url = infra.control_plane_url
     client = OrchestratorClient(url)
 
-    inputs = registration_inputs(egress_plan)
-    reg = client.register_bottle(
-        guest_ip, image_ref=image_ref, policy=inputs.policy,
-        metadata=inputs.metadata, tokens=tokens,
+    transport = infra_vm.gateway_transport()
+    reg = provision_bottle(
+        client, guest_ip, egress_plan, git_gate_plan, transport,
+        image_ref=image_ref, tokens=tokens,
     )
-    try:
-        provision_git_gate(
-            infra_vm.gateway_transport(), reg.bottle_id, git_gate_plan)
-    except Exception:
-        client.teardown_bottle(reg.bottle_id)
-        raise
-
     # The shared gateway CA every agent on this host trusts for TLS
     # interception — fetched from the infra VM over SSH.
     return LaunchContext(
@@ -88,6 +81,7 @@ def launch_consolidated(
         source_ip=guest_ip,
         gateway_ca_pem=infra.gateway_ca_pem(),
         orchestrator_url=url,
+        env_var_secret=reg.env_var_secret,
     )
 
 
@@ -98,12 +92,8 @@ def teardown_consolidated(
     VM. Both steps are idempotent so this is safe from a cleanup trap. Does
     NOT stop the infra VM — it's a persistent per-host singleton shared by
     every bottle."""
-    from ...orchestrator.config_store import DEFAULT_TEARDOWN_TIMEOUT_SECONDS
-    OrchestratorClient(
-        orchestrator_url,
-        timeout=timeout if timeout is not None else DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
-    ).teardown_bottle(bottle_id)
-    deprovision_git_gate(infra_vm.gateway_transport(), bottle_id)
+    _teardown_util(bottle_id, infra_vm.gateway_transport(),
+                   orchestrator_url=orchestrator_url, timeout=timeout)
 
 
 __all__ = [

bot_bottle/backend/firecracker/freezer.py

@@ -5,7 +5,7 @@ backend — we stream the guest root filesystem out over the control
 channel (SSH here). Unlike the other backends this needs no Docker: the
 tar *is* the resumable artifact. `resume` extracts it and rebuilds a
 fresh per-bottle ext4 with `mke2fs -d` (see `util.build_committed_rootfs_dir`
-and `launch._build_agent_base`). The bottle keeps running after the
+and `launch.build_or_load_agent_base`). The bottle keeps running after the
 snapshot.
 """
 

bot_bottle/backend/firecracker/image_builder.py

@@ -58,6 +58,12 @@ def _rootfs_digest(dockerfile: Path) -> str:
     return h.hexdigest()[:16]
 
 
+def cached_agent_rootfs_dir(dockerfile: Path) -> Path | None:
+    """Return the ready cached rootfs for ``dockerfile``, if one exists."""
+    base = util.cache_dir() / "rootfs" / f"agent-{_rootfs_digest(dockerfile)}"
+    return base if (base / ".bb-ready").is_file() else None
+
+
 def build_agent_rootfs_dir(
     dockerfile: Path, *, image_tag: str, smoke_test: tuple[str, ...] = (),
 ) -> Path:
@@ -72,7 +78,7 @@ def build_agent_rootfs_dir(
     silent-failure image at build time rather than at first agent use."""
     digest = _rootfs_digest(dockerfile)
     base = util.cache_dir() / "rootfs" / f"agent-{digest}"
-    if (base / ".bb-ready").is_file():
+    if cached_agent_rootfs_dir(dockerfile) is not None:
         info(f"using cached agent rootfs {base.name}")
         return base
 

bot_bottle/backend/firecracker/infra_artifact.py

@@ -41,7 +41,7 @@ from . import util
 _ARTIFACT_FORMAT = "1"
 
 _REPO_ROOT = Path(__file__).resolve().parents[3]
-_DOCKERFILES = ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra")
+_DOCKERFILES = ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra", "Dockerfile.infra.fc")
 
 _DEFAULT_BASE = "https://gitea.dideric.is"
 _DEFAULT_OWNER = "didericis"

bot_bottle/backend/firecracker/infra_vm.py

@@ -33,6 +33,7 @@ from pathlib import Path
 from typing import Generator
 
 from ...log import die, info
+from .. import util as backend_util
 from ..docker import util as docker_mod
 from ..docker.gateway_provision import GatewayProvisionError
 from . import firecracker_vm, infra_artifact, netpool, util
@@ -93,19 +94,18 @@ class InfraVm:
         """The gateway's mitmproxy CA (PEM) that agents install to trust its
         TLS interception. Generated a moment after boot, so this polls over
         SSH until it appears (mirrors DockerGateway.ca_cert_pem)."""
-        deadline = time.monotonic() + timeout
-        while True:
+        def _fetch() -> str | None:
             proc = subprocess.run(
                 util.ssh_base_argv(self.private_key, self.guest_ip)
                 + [f"cat {_GATEWAY_CA_PATH}"],
                 capture_output=True, text=True, timeout=15, check=False,
             )
-            if proc.returncode == 0 and "BEGIN CERTIFICATE" in proc.stdout:
-                return proc.stdout
-            if time.monotonic() >= deadline:
-                die(f"gateway CA not available after {timeout:g}s: "
-                    f"{proc.stderr.strip() or 'empty'}")
-            time.sleep(_HEALTH_POLL_SECONDS)
+            ok = proc.returncode == 0 and "BEGIN CERTIFICATE" in proc.stdout
+            return proc.stdout if ok else None
+        try:
+            return backend_util.poll_ca_cert(_fetch, timeout=timeout)
+        except TimeoutError as exc:
+            die(str(exc))
 
 
 def ensure_built() -> None:
@@ -125,16 +125,19 @@ def ensure_built() -> None:
 
 
 def build_infra_images_with_docker() -> None:
-    """Build the three fixed images from source with host Docker: orchestrator,
-    gateway, then the combined infra image (`COPY --from` orchestrator, `FROM`
-    gateway). The launch host uses this only in `BOT_BOTTLE_INFRA_BUILD=local`
-    mode; `publish_infra` uses it off-host to produce the published artifact."""
+    """Build the four fixed images from source with host Docker: orchestrator,
+    gateway, the shared infra base (Dockerfile.infra), then the Firecracker
+    infra image (Dockerfile.infra.fc: FROM infra + buildah). The launch host
+    uses this only in `BOT_BOTTLE_INFRA_BUILD=local` mode; `publish_infra`
+    uses it off-host to produce the published artifact."""
     docker_mod.build_image(
         _ORCHESTRATOR_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.orchestrator")
     docker_mod.build_image(
         _GATEWAY_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.gateway")
     docker_mod.build_image(
-        _INFRA_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.infra")
+        "bot-bottle-infra:latest", str(_REPO_ROOT), dockerfile="Dockerfile.infra")
+    docker_mod.build_image(
+        _INFRA_IMAGE, str(_REPO_ROOT), dockerfile="Dockerfile.infra.fc")
 
 
 def build_infra_rootfs_dir() -> Path:

bot_bottle/backend/firecracker/launch.py

@@ -45,7 +45,8 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale_path
+from ...log import die, info, warn
 from ...supervise import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
@@ -65,6 +66,7 @@ _GIT_HTTP_PORT = 9420
 @contextmanager
 def launch(
     plan: FirecrackerBottlePlan,
+    agent_base: Path,
     *,
     provision: Callable[[FirecrackerBottlePlan, "FirecrackerBottle"], str | None],
 ) -> Generator[FirecrackerBottle, None, None]:
@@ -86,11 +88,9 @@ def launch(
             raise teardown_exc
 
     try:
-        # Step 1: agent rootfs. Built from the Dockerfile inside a Firecracker
-        # builder VM (buildah, no host docker); a committed snapshot is reused
-        # when present. Returns the base dir the per-bottle ext4 is made from.
-        plan, agent_base = _build_agent_base(plan)
-
+        # Step 1 (rootfs resolution/build) runs in BottleBackend.launch before
+        # this context starts resources. ``agent_base`` is the selected cache,
+        # fresh build, or committed snapshot.
         # Step 2: mint the git-gate dynamic (gitea) deploy keys, if any.
         git_gate_plan = plan.git_gate_plan
         if git_gate_plan.upstreams:
@@ -209,9 +209,7 @@ def launch(
         teardown()
 
 
-def _build_agent_base(
-    plan: FirecrackerBottlePlan,
-) -> tuple[FirecrackerBottlePlan, Path]:
+def build_or_load_agent_base(plan: FirecrackerBottlePlan) -> Path:
     """Produce the agent's base rootfs dir. Primary path: build the Dockerfile
     inside a Firecracker builder VM (buildah, no host docker), smoke-testing
     the image before export. A committed snapshot (freeze/migrate) is resumed
@@ -220,13 +218,36 @@ def _build_agent_base(
     committed_tar = committed_rootfs_path(plan.slug)
     if committed and committed_tar.is_file():
         info(f"resuming from committed rootfs {committed_tar}")
-        return plan, util.build_committed_rootfs_dir(committed_tar)
-    base = image_builder.build_agent_rootfs_dir(
-        Path(plan.dockerfile_path),
+        return util.build_committed_rootfs_dir(committed_tar)
+    dockerfile = Path(plan.dockerfile_path)
+    if plan.spec.image_policy == "cached":
+        cached = image_builder.cached_agent_rootfs_dir(dockerfile)
+        if cached is None:
+            die(
+                f"cached agent rootfs for {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent rootfs {cached.name}")
+        return cached
+    return image_builder.build_agent_rootfs_dir(
+        dockerfile,
         image_tag=plan.image,
         smoke_test=runtime_for(plan.agent_provider_template).smoke_test,
     )
-    return plan, base
+
+
+def stale_checks(plan: FirecrackerBottlePlan) -> None:
+    """Raise when the cached rootfs selected by this plan is stale."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    committed_tar = committed_rootfs_path(plan.slug)
+    if committed and committed_tar.is_file():
+        check_stale_path(f"agent rootfs {committed_tar}", committed_tar)
+        return
+    cached = image_builder.cached_agent_rootfs_dir(Path(plan.dockerfile_path))
+    if cached is not None:
+        check_stale_path(f"agent rootfs {cached}", cached / ".bb-ready")
 
 
 # --- agent guest env -------------------------------------------------
@@ -242,6 +263,11 @@ def _agent_guest_env(plan: FirecrackerBottlePlan, host_ip: str) -> dict[str, str
         "HTTPS_PROXY": proxy_url, "HTTP_PROXY": proxy_url,
         "https_proxy": proxy_url, "http_proxy": proxy_url,
         "NO_PROXY": no_proxy, "no_proxy": no_proxy,
+        # Rootfs export can leave Git's implicit XDG paths unreadable even
+        # after the runtime repair. Bypass that discovery and name the
+        # provisioned global config explicitly so insteadOf can never fall
+        # through to the credential-bearing upstream URL.
+        "GIT_CONFIG_GLOBAL": f"{plan.guest_home}/.gitconfig",
         "NODE_EXTRA_CA_CERTS": AGENT_CA_PATH,
         "SSL_CERT_FILE": AGENT_CA_BUNDLE,
         "REQUESTS_CA_BUNDLE": AGENT_CA_BUNDLE,

bot_bottle/backend/firecracker/publish_infra.py

@@ -131,6 +131,29 @@ def build_artifact(out_dir: Path) -> tuple[str, Path, Path]:
     return version, gz, sha
 
 
+def _try_download_published(out_dir: Path) -> tuple[str, Path, Path] | None:
+    """If this version's artifact is already in the registry, download the gz
+    and sha to out_dir and return (version, gz_path, sha_path). Returns None
+    when not yet published."""
+    version = infra_artifact.infra_artifact_version(infra_vm._infra_init())
+    sha_url = infra_artifact.artifact_url(version, "rootfs.ext4.gz.sha256")
+    try:
+        with urllib.request.urlopen(infra_artifact._open(sha_url)):
+            pass
+    except urllib.error.HTTPError as e:
+        if e.code == 404:
+            return None
+        raise SystemExit(f"registry check failed (HTTP {e.code}): {sha_url}")
+    except urllib.error.URLError as e:
+        raise SystemExit(f"registry unreachable: {sha_url} ({e.reason})")
+    print(f"infra rootfs {version} already published — downloading instead of building")
+    gz = out_dir / "rootfs.ext4.gz"
+    sha = out_dir / "rootfs.ext4.gz.sha256"
+    infra_artifact._download(infra_artifact.artifact_url(version, "rootfs.ext4.gz"), gz)
+    infra_artifact._download(infra_artifact.artifact_url(version, "rootfs.ext4.gz.sha256"), sha)
+    return version, gz, sha
+
+
 def _publish_bundle(root: Path, token: str) -> str:
     version_file = root / "version.txt"
     # Guard the read so a missing version.txt is a clean error, not a raw
@@ -187,6 +210,8 @@ def main(argv: list[str] | None = None) -> int:
                       help="build a candidate bundle in DIR without publishing")
     mode.add_argument("--publish-dir", type=Path,
                       help="publish an already-built and tested candidate bundle")
+    parser.add_argument("--reuse-published", action="store_true",
+                        help="with --output: download from registry if already published instead of building")
     args = parser.parse_args(argv)
 
     _, _, token = infra_artifact._config()
@@ -197,6 +222,14 @@ def main(argv: list[str] | None = None) -> int:
 
     if args.output is not None:
         args.output.mkdir(parents=True, exist_ok=True)
+        reused = None
+        if args.reuse_published:
+            reused = _try_download_published(args.output)
+        if reused is not None:
+            version, _, _ = reused
+            (args.output / "version.txt").write_text(version + "\n", encoding="utf-8")
+            print(f"reused published infra rootfs candidate {version}")
+            return 0
         version, _gz, _sha = build_artifact(args.output)
         (args.output / "version.txt").write_text(version + "\n", encoding="utf-8")
         print(f"built infra rootfs candidate {version}")

bot_bottle/backend/firecracker/util.py

@@ -373,6 +373,12 @@ mount -o remount,rw / 2>/dev/null
 # scratch dirs there — git worktrees, build temp, `git init /tmp/...`, etc.
 mkdir -p /tmp && chmod 1777 /tmp
 
+# Rootfs export also maps the image's original owners to the unprivileged
+# host build uid. That uid is not guaranteed to be node's uid in the guest;
+# restore the home-directory boundary before any SSH provisioning runs.
+chown node:node /home/node 2>/dev/null || true
+chmod 755 /home/node 2>/dev/null || true
+
 # Install the per-bottle SSH pubkey from the kernel cmdline.
 KEY=$(sed -n 's/.*bb_pubkey=\([^ ]*\).*/\1/p' /proc/cmdline | base64 -d 2>/dev/null)
 if [ -n "$KEY" ]; then

bot_bottle/backend/macos_container/backend.py

@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
 from ...git_gate import GitGatePlan
 from ...supervise import SupervisePlan
 from ...manifest import Manifest
-from .. import ActiveAgent, BottleBackend, BottleSpec
+from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
 from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
@@ -82,11 +82,17 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def prelaunch_checks(self, plan: MacosContainerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
+    def _build_or_load_images(self, plan: MacosContainerBottlePlan) -> BottleImages:
+        return _launch.build_or_load_images(plan)
+
     @contextmanager
-    def launch(
-        self, plan: MacosContainerBottlePlan
+    def _launch_impl(
+        self, plan: MacosContainerBottlePlan, images: BottleImages
     ) -> Generator[MacosContainerBottle, None, None]:
-        with _launch.launch(plan, provision=self.provision) as bottle:
+        with _launch.launch(plan, images, provision=self.provision) as bottle:
             yield bottle
 
     def ensure_orchestrator(self) -> str:

bot_bottle/backend/macos_container/bottle_plan.py

@@ -20,7 +20,6 @@ class MacosContainerBottlePlan(BottlePlan):
     # bottle is registered. See launch.py's stamp for why it lives here and not
     # only in the exec-time proxy env.
     identity_token: str = ""
-    docker_access: bool = False
 
     @property
     def container_name(self) -> str:

bot_bottle/backend/macos_container/consolidated_launch.py

@@ -38,8 +38,7 @@ from ...egress import EgressPlan
 from ...git_gate import GitGatePlan
 from ...log import info
 from ...orchestrator.client import OrchestratorClient, OrchestratorClientError
-from ...orchestrator.registration import registration_inputs
-from ..docker.gateway_provision import deprovision_git_gate, provision_git_gate
+from ..consolidated_util import provision_bottle, teardown_consolidated as _teardown_util
 from . import util as container_mod
 from .enumerate import CONTAINER_NAME_PREFIX, EnumerationError, enumerate_active
 from .gateway import GATEWAY_NETWORK
@@ -73,6 +72,7 @@ class LaunchContext:
     gateway_ip: str
     network: str
     orchestrator_url: str
+    env_var_secret: str = ""  # encryption key injected into the agent's env
 
 
 def ensure_gateway(
@@ -142,17 +142,10 @@ def register_agent(
         client.reconcile(live_source_ips(endpoint.network))
     except (OrchestratorClientError, EnumerationError) as e:
         info(f"registry reconciliation skipped: {e}")
-    inputs = registration_inputs(egress_plan)
-    reg = client.register_bottle(
-        source_ip, image_ref=image_ref, policy=inputs.policy,
-        metadata=inputs.metadata, tokens=tokens,
+    reg = provision_bottle(
+        client, source_ip, egress_plan, git_gate_plan, AppleGatewayTransport(),
+        image_ref=image_ref, tokens=tokens,
     )
-    try:
-        provision_git_gate(AppleGatewayTransport(), reg.bottle_id, git_gate_plan)
-    except Exception:
-        # Roll the registration back so a provisioning failure leaves no orphan.
-        client.teardown_bottle(reg.bottle_id)
-        raise
     return LaunchContext(
         bottle_id=reg.bottle_id,
         identity_token=reg.identity_token,
@@ -160,6 +153,7 @@ def register_agent(
         gateway_ip=endpoint.gateway_ip,
         network=endpoint.network,
         orchestrator_url=endpoint.orchestrator_url,
+        env_var_secret=reg.env_var_secret,
     )
 
 
@@ -169,12 +163,8 @@ def teardown_consolidated(
     """Deregister the bottle and remove its git-gate state from the gateway.
     Both steps are idempotent so this is safe from a cleanup trap. Does NOT
     stop the gateway — it's a persistent per-host singleton."""
-    from ...orchestrator.config_store import DEFAULT_TEARDOWN_TIMEOUT_SECONDS
-    OrchestratorClient(
-        orchestrator_url,
-        timeout=timeout if timeout is not None else DEFAULT_TEARDOWN_TIMEOUT_SECONDS,
-    ).teardown_bottle(bottle_id)
-    deprovision_git_gate(AppleGatewayTransport(), bottle_id)
+    _teardown_util(bottle_id, AppleGatewayTransport(),
+                   orchestrator_url=orchestrator_url, timeout=timeout)
 
 
 __all__ = [

bot_bottle/backend/macos_container/infra.py

@@ -41,7 +41,7 @@ from dataclasses import dataclass
 from pathlib import Path
 
 from ... import log
-from ...orchestrator.gateway import GATEWAY_CA_CERT
+from ...orchestrator.gateway import GATEWAY_CA_CERT, MITMPROXY_HOME
 from ...orchestrator.lifecycle import (
     DEFAULT_PORT,
     DEFAULT_STARTUP_TIMEOUT_SECONDS,
@@ -52,7 +52,9 @@ from ...paths import (
     CONTROL_PLANE_TOKEN_ENV,
     HOST_DB_FILENAME,
     host_control_plane_token,
+    host_gateway_ca_dir,
 )
+from .. import util as backend_util
 from . import util as container_mod
 from .gateway import (
     DEFAULT_CA_TIMEOUT_SECONDS,
@@ -217,6 +219,14 @@ class MacosInfraService:
             # Container-only DB volume: one kernel writes bot-bottle.db, never
             # shared with the host or another guest.
             "--volume", f"{self._db_volume}:{_DB_ROOT_IN_CONTAINER}",
+            # The DB needs a container-only ext4 volume for coherent SQLite
+            # locking, but the CA has no such constraint. Keep it in the host
+            # app-data root so infra-container recreation and Apple Container
+            # volume pruning cannot silently rotate every bottle's trust
+            # anchor (issue #450).
+            "--mount",
+            container_mod.bind_mount_spec(
+                str(host_gateway_ca_dir()), MITMPROXY_HOME),
             # Bind-mount the control-plane source (read-only); a code change
             # takes effect on relaunch with no image rebuild.
             "--mount",
@@ -260,21 +270,19 @@ class MacosInfraService:
 
     def ca_cert_pem(self, *, timeout: float = DEFAULT_CA_TIMEOUT_SECONDS) -> str:
         """The gateway's mitmproxy CA (PEM) agents install to trust its TLS
-        interception. Read out of the container (the CA lives on a
-        container-internal path, not a host mount); polls because mitmproxy
-        writes it a beat after start."""
-        deadline = time.monotonic() + timeout
-        while True:
+        interception. Read through the container path backed by the persistent
+        host CA directory; polls because mitmproxy writes it a beat after
+        start."""
+        def _fetch() -> str | None:
             result = container_mod.run_container_argv(
                 ["container", "exec", self._name, "cat", GATEWAY_CA_CERT])
-            if result.returncode == 0 and result.stdout.strip():
-                return result.stdout
-            if time.monotonic() >= deadline:
-                raise GatewayError(
-                    f"gateway CA not available in {self._name} after {timeout:g}s: "
-                    f"{(result.stderr or '').strip() or 'empty'}"
-                )
-            time.sleep(_CA_POLL_SECONDS)
+            return result.stdout if result.returncode == 0 and result.stdout.strip() else None
+        try:
+            return backend_util.poll_ca_cert(_fetch, timeout=timeout)
+        except TimeoutError as exc:
+            raise GatewayError(
+                f"gateway CA not available in {self._name} after {timeout:g}s"
+            ) from exc
 
     def stop(self) -> None:
         """Remove the infra container (idempotent). The DB volume persists."""

bot_bottle/backend/macos_container/launch.py

@@ -53,7 +53,9 @@ from ...git_gate import (
     revoke_git_gate_provisioned_keys,
 )
 from ...git_http_backend import DEFAULT_PORT as _GIT_HTTP_PORT
+from ...image_cache import check_stale
 from ...log import die, info, warn
+from .. import BottleImages
 from ...supervise import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
@@ -64,7 +66,6 @@ from .gateway_hosts import (
     refresh_gateway_host,
     set_gateway_host,
 )
-from . import rootless_podman
 from .bottle_plan import MacosContainerBottlePlan
 from ...orchestrator.config_store import resolve_teardown_timeout
 from .consolidated_launch import (
@@ -78,18 +79,43 @@ _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
 _AGENT_SLEEP_SECONDS = "2147483647"
 
 
+def build_or_load_images(plan: MacosContainerBottlePlan) -> BottleImages:
+    """Resolve the agent image ref for this plan. The gateway's own image is
+    built by `ensure_gateway` — it belongs to the shared singleton."""
+    committed = read_committed_image(plan.slug)
+    if committed and container_mod.image_exists(committed):
+        info(f"using committed image {committed!r}")
+        return BottleImages(agent=committed)
+    if plan.spec.image_policy == "cached":
+        if not container_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
+            )
+        info(f"using cached agent image {plan.image!r}")
+        return BottleImages(agent=plan.image)
+    container_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
+    return BottleImages(agent=plan.image)
+
+
 @contextmanager
 def launch(
     plan: MacosContainerBottlePlan,
+    images: BottleImages,
     *,
     provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
 ) -> Generator[MacosContainerBottle, None, None]:
-    """Build, run, register, provision, and yield an Apple Container bottle on
-    the shared per-host gateway."""
+    """Run, register, provision, and yield an Apple Container bottle on the
+    shared per-host gateway."""
     stack = ExitStack()
     bottle_for_revoke = plan.manifest.bottle
     git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
 
+    plan = dataclasses.replace(
+        plan,
+        agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
+    )
+
     def teardown() -> None:
         teardown_exc: BaseException | None = None
         try:
@@ -102,8 +128,6 @@ def launch(
             raise teardown_exc
 
     try:
-        plan = _build_images(plan)
-
         # Step 1: the per-host singletons. Must precede the agent run — its
         # proxy env needs the gateway's address at `container run` time.
         endpoint = ensure_gateway()
@@ -172,10 +196,6 @@ def launch(
         # token above, so — unlike the run-time env — the plan CAN carry it.
         plan = dataclasses.replace(plan, identity_token=ctx.identity_token)
 
-        exec_env = {
-            **_identity_proxy_env(endpoint, ctx.identity_token),
-            **rootless_podman.guest_env(plan.docker_access),
-        }
         bottle = MacosContainerBottle(
             plan.container_name,
             teardown,
@@ -189,44 +209,32 @@ def launch(
             ),
             terminal_color=plan.spec.color,
             agent_workdir=plan.workspace_plan.workdir,
-            exec_env=exec_env,
+            exec_env=_identity_proxy_env(endpoint, ctx.identity_token),
         )
         bottle.prompt_path = provision(plan, bottle)
 
-        if plan.docker_access:
-            rootless_podman.prepare_guest_devices(
-                plan.container_name, container_mod.exec_container_as_root,
-            )
-            rootless_podman.start(bottle)
-
         yield bottle
     finally:
         teardown()
 
 
-def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
-    """Build the agent image. The gateway's own image is built by
-    `ensure_gateway` — it belongs to the shared singleton, not to a bottle."""
+
+def stale_checks(plan: MacosContainerBottlePlan) -> None:
+    """Raise StaleImageError if a cached image is older than the configured
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
     committed = read_committed_image(plan.slug)
     if committed and container_mod.image_exists(committed):
-        info(f"using committed image {committed!r}")
-        plan = dataclasses.replace(
-            plan,
-            agent_provision=dataclasses.replace(
-                plan.agent_provision, image=committed,
-            ),
-        )
-    else:
-        container_mod.build_image(
-            plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path,
-        )
-    if plan.docker_access:
-        image = rootless_podman.build_image(plan.image, container_mod.build_image)
-        plan = dataclasses.replace(
-            plan,
-            agent_provision=dataclasses.replace(plan.agent_provision, image=image),
-        )
-    return plan
+        ts = container_mod.image_created_at(committed)
+        if ts is not None:
+            check_stale(f"agent image {committed!r}", ts)
+        return
+    if container_mod.image_exists(plan.image):
+        ts = container_mod.image_created_at(plan.image)
+        if ts is not None:
+            check_stale(f"agent image {plan.image!r}", ts)
 
 
 def _provision_git_gate_keys(

bot_bottle/backend/macos_container/resolve_plan.py

@@ -44,5 +44,4 @@ def resolve_plan(
         egress_plan=egress_plan,
         supervise_plan=supervise_plan,
         agent_provision=agent_provision_plan,
-        docker_access=manifest.bottle.docker_access,
     )

bot_bottle/backend/macos_container/rootless-podman-init.sh

@@ -1,89 +0,0 @@
-#!/bin/sh
-set -eu
-
-uid="$(id -u)"
-if [ "$uid" -eq 0 ]; then
-    echo "refusing to run rootless podman as root" >&2
-    exit 1
-fi
-
-for command in podman docker fuse-overlayfs slirp4netns; do
-    command -v "$command" >/dev/null 2>&1 || {
-        echo "missing rootless podman prerequisite: $command" >&2
-        exit 1
-    }
-done
-
-# The inverse of the rootless-Docker check, and the whole point of the podman
-# variant: a subordinate range would push podman onto newuidmap, which cannot
-# write a multi-range uid_map without CAP_SYS_ADMIN in this guest.  An empty
-# range keeps it on the single-UID self-mapping an unprivileged process may
-# write itself.
-if grep -q "^$(id -un):" /etc/subuid 2>/dev/null; then
-    echo "unexpected subordinate UID range for $(id -un): podman would" >&2
-    echo "require CAP_SYS_ADMIN via newuidmap in this guest" >&2
-    exit 1
-fi
-
-for device in /dev/fuse /dev/net/tun; do
-    [ -r "$device" ] && [ -w "$device" ] || {
-        echo "device $device is not readable/writable by $(id -un)" >&2
-        exit 1
-    }
-done
-
-export XDG_RUNTIME_DIR="${XDG_RUNTIME_DIR:-/tmp/bot-bottle-podman-run}"
-config="$HOME/.config/containers"
-mkdir -p "$XDG_RUNTIME_DIR" "$config"
-chmod 700 "$XDG_RUNTIME_DIR"
-
-# ignore_chown_errors is required, not incidental: with a single-UID mapping
-# there is no second UID for image layers to be chowned to, so layers that
-# record other owners would otherwise fail to extract.
-cat > "$config/storage.conf" <<'CONF'
-[storage]
-driver="overlay"
-[storage.options.overlay]
-mount_program="/usr/bin/fuse-overlayfs"
-ignore_chown_errors="true"
-CONF
-
-# No cgroup delegation reaches this guest, so asking podman to manage cgroups
-# fails; events_logger=file avoids the journald socket that is equally absent.
-cat > "$config/containers.conf" <<'CONF'
-[containers]
-cgroups="disabled"
-[engine]
-cgroup_manager="cgroupfs"
-events_logger="file"
-CONF
-
-# Registry pulls egress through the bottle's proxy like everything else. The
-# token-bearing proxy URL is already in the agent's environment; persisting it
-# inside this disposable VM does not broaden its authority.
-python3 - <<'PY'
-import json
-import os
-from pathlib import Path
-
-proxy = os.environ.get("HTTPS_PROXY") or os.environ.get("https_proxy", "")
-no_proxy = os.environ.get("NO_PROXY") or os.environ.get("no_proxy", "")
-config = {"proxies": {"default": {
-    "httpProxy": proxy,
-    "httpsProxy": proxy,
-    "noProxy": no_proxy,
-}}}
-path = Path.home() / ".docker" / "config.json"
-path.parent.mkdir(parents=True, exist_ok=True)
-path.write_text(json.dumps(config), encoding="utf-8")
-path.chmod(0o600)
-PY
-
-if docker info >/dev/null 2>&1; then
-    exit 0
-fi
-
-log=/tmp/bot-bottle-rootless-podman.log
-nohup podman system service --time=0 \
-    "unix://$XDG_RUNTIME_DIR/podman.sock" \
-    >"$log" 2>&1 </dev/null &

bot_bottle/backend/macos_container/rootless_podman.py

@@ -1,132 +0,0 @@
-"""Experimental rootless podman bootstrap for Apple-container bottles.
-
-The service and every nested container remain inside the existing per-bottle
-VM.  This module refuses to compensate for missing prerequisites with outer
-capabilities, a privileged container, or a host Docker socket.
-
-Podman is used rather than rootless Docker for one specific reason: Apple
-Container's capability bounding set omits `CAP_SYS_ADMIN`, which the kernel
-requires to write a multi-range `uid_map` via `newuidmap`.  Rootless Docker
-has no path that avoids that write.  Podman does — with no subordinate UID
-range configured it falls back to a single-UID self-mapping, which an
-unprivileged process may write itself.  See
-`docs/research/rootless-docker-in-apple-container-spike.md`.
-
-That fallback is why `build_image` *removes* the agent user's `/etc/subuid`
-and `/etc/subgid` entries instead of adding them: their presence is precisely
-what would send podman down the `newuidmap` path that cannot work here.
-
-The agent still talks to `docker` and `docker compose`; those speak to
-podman's Docker-compatible API socket, so nothing in the agent's habits
-changes.
-"""
-
-from __future__ import annotations
-
-import shlex
-import shutil
-import tempfile
-import time
-from pathlib import Path
-from typing import Callable
-
-from ...log import die, info
-
-_INIT = "/usr/local/libexec/bot-bottle/rootless-podman-init"
-_RUNTIME_DIR = "/tmp/bot-bottle-podman-run"
-_SOCKET = f"{_RUNTIME_DIR}/podman.sock"
-_LOG = "/tmp/bot-bottle-rootless-podman.log"
-READY_RETRIES = 30
-
-# Apple Container creates both device nodes 0600 root:root, so the agent user
-# cannot open them: /dev/fuse blocks the fuse-overlayfs storage driver and
-# /dev/net/tun blocks slirp4netns, which rootless podman uses for the default
-# bridge network that stock compose files expect.  Relaxing the modes needs no
-# capability the bottle does not already hold — unlike CAP_SYS_ADMIN, which is
-# what killed the rootless-Docker approach.
-_GUEST_DEVICES = ("/dev/fuse", "/dev/net/tun")
-
-
-def build_image(
-    base_image: str,
-    build: Callable[..., None],
-) -> str:
-    """Layer spike-only tooling on an already-built provider image."""
-    image = f"{base_image}-rootless-podman"
-    init_script = Path(__file__).with_name("rootless-podman-init.sh")
-    with tempfile.TemporaryDirectory(prefix="bot-bottle-rootless-podman.") as tmp:
-        context = Path(tmp)
-        shutil.copy2(init_script, context / "rootless-podman-init.sh")
-        (context / "Dockerfile").write_text(
-            "FROM docker:28-cli AS docker_cli\n"
-            f"FROM {base_image}\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/"
-            "docker-compose /usr/local/libexec/docker/cli-plugins/docker-compose\n"
-            "RUN apt-get update \\\n"
-            " && apt-get install -y --no-install-recommends podman "
-            "fuse-overlayfs slirp4netns uidmap \\\n"
-            " && rm -rf /var/lib/apt/lists/* \\\n"
-            # Deliberate: an empty subordinate range keeps podman on the
-            # single-UID mapping that needs no CAP_SYS_ADMIN.  Adding ranges
-            # here would reintroduce the newuidmap failure this spike exists
-            # to route around.
-            " && sed -i '/^node:/d' /etc/subuid /etc/subgid\n"
-            "COPY rootless-podman-init.sh "
-            "/usr/local/libexec/bot-bottle/rootless-podman-init\n"
-            "RUN chmod 0755 /usr/local/libexec/bot-bottle/rootless-podman-init\n"
-            "USER node\n",
-            encoding="utf-8",
-        )
-        build(image, str(context), dockerfile=str(context / "Dockerfile"))
-    return image
-
-
-def guest_env(enabled: bool) -> dict[str, str]:
-    """Environment consumed by the Docker CLI inside an enabled bottle."""
-    if not enabled:
-        return {}
-    return {
-        "DOCKER_HOST": f"unix://{_SOCKET}",
-        "XDG_RUNTIME_DIR": _RUNTIME_DIR,
-    }
-
-
-def prepare_guest_devices(container_name: str, exec_as_root: Callable[..., None]) -> None:
-    """Make /dev/fuse and /dev/net/tun openable by the agent user.
-
-    Runs as root inside the bottle because the agent must not be able to
-    re-mode device nodes itself.  No outer capability is involved.
-    """
-    exec_as_root(
-        container_name,
-        ["sh", "-c", f"chmod 0666 {' '.join(_GUEST_DEVICES)}"],
-    )
-
-
-def start(bottle: object) -> None:
-    """Start and verify the unprivileged service through the bottle exec API."""
-    info("starting experimental rootless podman service")
-    result = bottle.exec(shlex.quote(_INIT))  # type: ignore[attr-defined]
-    if result.returncode != 0:
-        detail = (result.stderr or result.stdout or "").strip()
-        die(f"rootless podman bootstrap failed: {detail or '<no output>'}")
-
-    for _ in range(READY_RETRIES):
-        result = bottle.exec("docker info >/dev/null 2>&1")  # type: ignore[attr-defined]
-        if result.returncode == 0:
-            info("rootless podman service is ready")
-            return
-        time.sleep(0.2)
-
-    logs = bottle.exec(  # type: ignore[attr-defined]
-        f"tail -n 80 {_LOG} 2>/dev/null || true"
-    )
-    die(
-        "rootless podman did not become ready without additional outer "
-        f"privileges:\n{(logs.stdout or logs.stderr or '<no log>').strip()}"
-    )
-
-
-__all__ = ["build_image", "guest_env", "prepare_guest_devices", "start"]

bot_bottle/backend/macos_container/util.py

@@ -10,6 +10,7 @@ import shutil
 import subprocess
 import tempfile
 import time
+from datetime import datetime, timezone
 from typing import Iterable
 
 from ...log import die, info
@@ -661,6 +662,39 @@ def image_id(ref: str) -> str:
     raise AssertionError("unreachable")
 
 
+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
+    pulled from registries that omit the field). Callers should skip the stale
+    check when None is returned rather than treating it as an error."""
+    result = subprocess.run(
+        [_CONTAINER, "image", "inspect", ref],
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+    if result.returncode != 0:
+        die(
+            f"container image inspect for {ref!r} failed: "
+            f"{(result.stderr or '').strip() or '<no stderr>'}"
+        )
+    try:
+        data = json.loads(result.stdout or "{}")
+    except json.JSONDecodeError as exc:
+        die(f"container image inspect for {ref!r} returned malformed JSON: {exc}")
+    if isinstance(data, list) and data:
+        data = data[0]
+    if isinstance(data, dict):
+        value = data.get("created") or data.get("Created")
+        if isinstance(value, str) and value:
+            try:
+                ts = value.rstrip("Z")
+                return datetime.fromisoformat(ts).replace(tzinfo=timezone.utc)
+            except ValueError:
+                pass
+    return None
+
+
 def save(ref: str, output: str) -> None:
     subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True)
 

bot_bottle/backend/util.py

@@ -7,6 +7,8 @@ from __future__ import annotations
 import hashlib
 import os
 import ssl
+import time
+from collections.abc import Callable
 from pathlib import Path
 from typing import TYPE_CHECKING
 
@@ -15,6 +17,24 @@ from ..log import die, info
 if TYPE_CHECKING:
     from ..egress import EgressPlan
 
+_CA_POLL_INTERVAL = 0.5
+
+
+def poll_ca_cert(fetch: Callable[[], str | None], *, timeout: float) -> str:
+    """Poll `fetch` until it returns a non-empty PEM string or `timeout` expires.
+
+    `fetch` should return the PEM on success and `None` (or empty string) when
+    the cert is not yet available. Raises `TimeoutError` if the cert never
+    appears within `timeout` seconds."""
+    deadline = time.monotonic() + timeout
+    while True:
+        result = fetch()
+        if result:
+            return result
+        if time.monotonic() >= deadline:
+            raise TimeoutError(f"CA cert not available after {timeout:g}s")
+        time.sleep(_CA_POLL_INTERVAL)
+
 
 # Debian-family CA layout, shared by every backend (all guest images
 # are Debian-family). AGENT_CA_PATH is the source path that

bot_bottle/cli/__init__.py

@@ -19,6 +19,7 @@ from .commit import cmd_commit
 from .edit import cmd_edit
 from .info import cmd_info
 from .init import cmd_init
+from .login import cmd_login
 from .resume import cmd_resume
 from .start import cmd_start
 from .supervise import cmd_supervise
@@ -33,6 +34,7 @@ COMMANDS = {
     "info": cmd_info,
     "init": cmd_init,
     "list": cmd_list,
+    "login": cmd_login,
     "resume": cmd_resume,
     "start": cmd_start,
     "supervise": cmd_supervise,
@@ -43,7 +45,7 @@ COMMANDS = {
 # the host (TAP pool, /dev/kvm, firecracker) and never opens the store, so
 # gating it on the schema breaks preflight on a fresh CI runner where stdin
 # isn't a TTY and the migration prompt can't be answered.
-NO_MIGRATION_COMMANDS = frozenset({"backend"})
+NO_MIGRATION_COMMANDS = frozenset({"backend", "login"})
 
 
 def usage() -> None:
@@ -56,6 +58,7 @@ def usage() -> None:
     sys.stderr.write("  info      print env, skills, and prompt details for a named agent\n")
     sys.stderr.write("  init      interactively create a new agent and add it to bot-bottle.json\n")
     sys.stderr.write("  list      list available agents or active containers\n")
+    sys.stderr.write("  login     register this host with a bot-bottle console\n")
     sys.stderr.write(
         "  resume    re-launch a bottle by its identity "
         "(continues state from PRD 0016)\n"
@@ -111,7 +114,3 @@ def main(argv: list[str] | None = None) -> int:
         return e.code if isinstance(e.code, int) else 1
     except KeyboardInterrupt:
         return 130
-
-
-if __name__ == "__main__":
-    sys.exit(main())

bot_bottle/cli/__main__.py

@@ -0,0 +1,15 @@
+"""Entry point for `python -m bot_bottle.cli`.
+
+`cli.py` at the repo root is the usual way in; this makes the package
+runnable too, so the CLI works from an installed copy where there is no
+`cli.py` on disk to point at.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from . import main
+
+if __name__ == "__main__":
+    sys.exit(main())

bot_bottle/cli/login.py

@@ -0,0 +1,168 @@
+"""bb login — register this host with a bot-bottle console.
+
+Opens a device-authorization flow against the target console, waits for the
+operator to approve, then writes access and refresh tokens to
+~/.bot-bottle/console.json (or $BOT_BOTTLE_ROOT/console.json).
+
+Usage:
+  bb login [--console-url URL] [--label LABEL]
+
+Flags:
+  --console-url URL   Target console URL (overrides BB_CONSOLE_URL env var)
+  --label LABEL       Host label shown in the console (default: hostname)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+from ..paths import bot_bottle_root
+
+_CONSOLE_URL_ENV = "BB_CONSOLE_URL"
+_POLL_SLEEP = 2  # seconds between polls; matches console's poll_interval default
+
+
+def _usage() -> None:
+    sys.stderr.write(
+        "usage: bb login [--console-url URL] [--label LABEL]\n"
+        "\n"
+        "Options:\n"
+        "  --console-url URL  Console base URL (or BB_CONSOLE_URL env var)\n"
+        "  --label LABEL      Host label shown in the console (default: hostname)\n"
+    )
+
+
+def _flag(argv: list[str], name: str) -> str | None:
+    for i, arg in enumerate(argv):
+        if arg == name and i + 1 < len(argv):
+            return argv[i + 1]
+        if arg.startswith(f"{name}="):
+            return arg[len(name) + 1:]
+    return None
+
+
+def _post(url: str, payload: dict[str, Any]) -> dict[str, Any]:
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(
+        url, data=data, headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        return json.loads(resp.read())
+
+
+def _get(url: str) -> tuple[int, dict[str, Any]]:
+    req = urllib.request.Request(url)
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return resp.status, json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        return e.code, {}
+
+
+def _save_credentials(
+    console_url: str, host_id: str, access_token: str, refresh_token: str
+) -> Path:
+    path = bot_bottle_root() / "console.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = (
+        json.dumps(
+            {
+                "url": console_url,
+                "host_id": host_id,
+                "access_token": access_token,
+                "refresh_token": refresh_token,
+            },
+            indent=2,
+        )
+        + "\n"
+    )
+    fd, tmp_path_str = tempfile.mkstemp(dir=path.parent, prefix=".console-")
+    tmp = Path(tmp_path_str)
+    try:
+        tmp.chmod(0o600)
+        with os.fdopen(fd, "w") as f:
+            f.write(content)
+        os.replace(tmp, path)
+    except OSError:
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+        raise
+    return path
+
+
+def cmd_login(argv: list[str]) -> int:
+    if "--help" in argv or "-h" in argv:
+        _usage()
+        return 0
+
+    console_url = _flag(argv, "--console-url") or os.environ.get(_CONSOLE_URL_ENV)
+    if not console_url:
+        sys.stderr.write(
+            "bb login: --console-url or BB_CONSOLE_URL is required\n"
+        )
+        return 1
+    console_url = console_url.rstrip("/")
+
+    label = _flag(argv, "--label") or socket.gethostname()
+
+    try:
+        resp = _post(f"{console_url}/api/v1/hosts/authorize", {"label": label})
+    except (OSError, ValueError) as exc:
+        sys.stderr.write(f"bb login: failed to start authorization: {exc}\n")
+        return 1
+
+    device_code = resp["device_code"]
+    user_code = resp["user_code"]
+    expires_in = resp.get("expires_in", 300)
+    poll_sleep = max(1, min(int(resp.get("poll_interval", _POLL_SLEEP)), 60))
+
+    sys.stderr.write(
+        f"\nOpen this URL in your browser to authorize this host:\n\n"
+        f"  {console_url}/hosts/authorize?code={user_code}\n\n"
+        f"Waiting for approval"
+    )
+
+    deadline = time.monotonic() + expires_in
+    while time.monotonic() < deadline:
+        sys.stderr.write(".")
+        sys.stderr.flush()
+        time.sleep(poll_sleep)
+
+        try:
+            code, result = _get(
+                f"{console_url}/api/v1/hosts/authorize/{device_code}"
+            )
+        except (OSError, ValueError):
+            continue
+
+        if code == 410:
+            break
+
+        st = result.get("status")
+        if st == "approved":
+            sys.stderr.write("\n\nApproved.\n")
+            path = _save_credentials(
+                console_url,
+                result["host_id"],
+                result["access_token"],
+                result["refresh_token"],
+            )
+            sys.stderr.write(f"Credentials saved to {path}\n")
+            return 0
+        if st == "denied":
+            sys.stderr.write("\n\nDenied by operator.\n")
+            return 1
+
+    sys.stderr.write("\n\nAuthorization timed out.\n")
+    return 1

bot_bottle/cli/start.py

@@ -35,6 +35,7 @@ from ..bottle_state import (
     is_preserved,
     mark_preserved,
 )
+from ..image_cache import StaleImageError
 from ..log import info, die
 from ..manifest import Manifest, ManifestIndex
 from ._common import PROG, USER_CWD, read_tty_line
@@ -64,6 +65,14 @@ def cmd_start(argv: list[str]) -> int:
             "skip all prompts. For orchestrators, CI, and webhooks."
         ),
     )
+    parser.add_argument(
+        "--cached-images",
+        action="store_true",
+        help=(
+            "quickstart with existing local agent and sidecar images; "
+            "only valid with --headless"
+        ),
+    )
     parser.add_argument(
         "--bottle",
         action="append",
@@ -96,6 +105,8 @@ def cmd_start(argv: list[str]) -> int:
         help="agent name defined in bot-bottle.json (omit to pick interactively)",
     )
     args = parser.parse_args(argv)
+    if args.cached_images and not args.headless:
+        die("--cached-images is only supported with --headless")
 
     dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
     if args.no_cache or os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
@@ -147,6 +158,10 @@ def cmd_start(argv: list[str]) -> int:
     label, color = tui.name_color_modal(default_label=agent_name)
     label, color = _resolve_unique_label(label, color)
 
+    image_policy = _select_image_policy()
+    if image_policy is None:
+        return 0
+
     spec = BottleSpec(
         manifest=manifest,
         agent_name=agent_name,
@@ -155,6 +170,7 @@ def cmd_start(argv: list[str]) -> int:
         label=label,
         color=color,
         bottle_names=bottle_names,
+        image_policy=image_policy,
     )
     return _launch_bottle(
         spec,
@@ -213,6 +229,7 @@ def _start_headless(
         color=args.color or "",
         bottle_names=bottle_names,
         headless=True,
+        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -395,6 +412,13 @@ def _text_prompt_yes() -> bool:
     return reply in ("y", "Y", "yes", "YES")
 
 
+def _select_image_policy() -> str | None:
+    return tui.filter_select(
+        ["fresh", "cached"],
+        title="Select image startup mode",
+    )
+
+
 def _text_render_preflight():
     def _render(plan: DockerBottlePlan, backend_name: str) -> None:
         print(file=sys.stderr)
@@ -537,6 +561,15 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
+        try:
+            backend.prelaunch_checks(plan)
+        except StaleImageError as exc:
+            if assume_yes:
+                die(str(exc))
+            sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
+            sys.stderr.flush()
+            if read_tty_line() not in ("y", "Y", "yes", "YES"):
+                return 0
         with backend.launch(plan) as bottle:
             agent_provider_template = getattr(plan, "agent_provider_template", "claude")
             extra_args: tuple[str, ...] = ()
@@ -555,10 +588,6 @@ def _launch_bottle(
                 f"session ended (exit {exit_code}); "
                 f"container {bottle.name} will be removed"
             )
-            # While the container is still alive: always snapshot the
-            # transcript and — if the agent exited non-zero — mark
-            # the state for preservation. This picks up crashes /
-            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
             if agent_provider_template == "claude":
                 capture_claude_session_state(identity, exit_code)
         return 0

bot_bottle/config_store.py

@@ -0,0 +1,71 @@
+"""SQLite-backed bot-bottle configuration store."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+try:
+    from .db_store import DbStore
+    from .migrations import TableMigrations
+    from .paths import host_db_path
+except ImportError:
+    from db_store import DbStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from migrations import TableMigrations  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from paths import host_db_path  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS = 1
+
+
+class ConfigStore(DbStore):
+    """SQLite configuration for host-side bot-bottle settings."""
+
+    def __init__(self, db_path: Path | None = None) -> None:
+        migrations = TableMigrations("config_store", [
+            # v1 — host-side bot-bottle settings
+            """
+            CREATE TABLE IF NOT EXISTS bot_bottle_config (
+                id INTEGER PRIMARY KEY CHECK (id = 1),
+                cached_image_stale_warning_days INTEGER NOT NULL DEFAULT 1
+            )
+            """,
+        ])
+        super().__init__(db_path or host_db_path(), migrations)
+
+    def cached_image_stale_warning_days(self) -> int:
+        if not self.db_path.is_file():
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+        with self._connect() as conn:
+            row = conn.execute(
+                """
+                SELECT cached_image_stale_warning_days
+                FROM bot_bottle_config
+                WHERE id = 1
+                """,
+            ).fetchone()
+        if row is None:
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+        try:
+            return int(row["cached_image_stale_warning_days"])
+        except (TypeError, ValueError):
+            return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
+
+    def set_cached_image_stale_warning_days(self, days: int) -> Path:
+        with self._connect() as conn:
+            conn.execute(
+                """
+                INSERT INTO bot_bottle_config (id, cached_image_stale_warning_days)
+                VALUES (1, ?)
+                ON CONFLICT(id) DO UPDATE SET
+                    cached_image_stale_warning_days = excluded.cached_image_stale_warning_days
+                """,
+                (days,),
+            )
+        self._chmod()
+        return self.db_path
+
+
+__all__ = [
+    "DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS",
+    "ConfigStore",
+]

bot_bottle/contrib/claude/Dockerfile

@@ -10,7 +10,7 @@
 
 # Current Node LTS; slim variant keeps the image small while still
 # providing apt-get for any future additions.
-FROM node:22-slim
+FROM node:22-trixie-slim
 
 # Install runtime system deps. claude-code shells out to git for several
 # features (status checks, commits, PR creation) — without git in the
@@ -21,7 +21,15 @@ FROM node:22-slim
 # to it) works against egress's bumped TLS without the agent needing
 # local DNS.
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils \
+  && apt-get install -y --no-install-recommends \
+  git \
+  ca-certificates \
+  curl \
+  openssh-client \
+  podman \
+  ripgrep \
+  iproute2 \
+  dnsutils \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by claude-code itself
@@ -39,6 +47,11 @@ RUN apt-get update \
 RUN npm install -g --no-fund --no-audit @anthropic-ai/claude-code@2.1.172 \
   && npm cache clean --force
 
+# Git reads both ~/.gitconfig and ~/.config/git/config. Keep its XDG config
+# path traversable by the non-root runtime user so permission errors do not
+# suppress bot-bottle's git-gate insteadOf rules.
+RUN install -d -o node -g node -m 755 /home/node/.config /home/node/.config/git
+
 # Run as a non-root user. The node image already provides a `node` user
 # (uid 1000) with a home directory, which is where claude-code will write
 # its session state.

bot_bottle/contrib/claude/agent_provider.py

@@ -23,8 +23,9 @@ from ...agent_provider import (
     provider_startup_args,
 )
 from ...backend.docker import util as docker_mod
-from ...egress import EgressRoute
+from ...egress import CLAUDE_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
 from ...log import die, info, warn
+from .claude_auth import claude_host_access_token
 
 
 if TYPE_CHECKING:
@@ -118,7 +119,6 @@ class ClaudeAgentProvider(AgentProvider):
         color: str = "",
         provider_settings: dict[str, object] | None = None,
     ) -> AgentProvisionPlan:
-        del forward_host_credentials, host_env
         resolved_guest_env = dict(guest_env or {})
         startup_args = provider_startup_args(provider_settings)
         guest_home = self.guest_home
@@ -180,13 +180,24 @@ class ClaudeAgentProvider(AgentProvider):
                 claude_settings,
                 f"{guest_home}/.claude/settings.json",
             ))
+        provisioned_env: dict[str, str] = {}
+        if forward_host_credentials:
+            _host_env = host_env or dict(os.environ)
+            provisioned_env[CLAUDE_HOST_CREDENTIAL_TOKEN_REF] = (
+                claude_host_access_token(_host_env)
+            )
+
+        cred_token_ref = (
+            CLAUDE_HOST_CREDENTIAL_TOKEN_REF if forward_host_credentials
+            else auth_token
+        )
         egress_routes = (EgressRoute(
             host="api.anthropic.com",
-            auth_scheme="Bearer" if auth_token else "",
-            token_ref=auth_token,
+            auth_scheme="Bearer" if (auth_token or forward_host_credentials) else "",
+            token_ref=cred_token_ref,
         ),)
         hidden_env_names: frozenset[str] = frozenset()
-        if auth_token:
+        if auth_token or forward_host_credentials:
             env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
             hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
 
@@ -208,6 +219,7 @@ class ClaudeAgentProvider(AgentProvider):
             files=tuple(files),
             egress_routes=egress_routes,
             hidden_env_names=hidden_env_names,
+            provisioned_env=provisioned_env,
         )
 
     def provision_skills(self, plan: "BottlePlan", bottle: "Bottle") -> None:

bot_bottle/contrib/claude/claude_auth.py

@@ -0,0 +1,114 @@
+"""Host Claude auth helpers.
+
+Reads the host's Claude Code credentials and returns only the access
+token needed by egress. Does not expose refresh tokens or raw payloads.
+
+Credential storage by platform:
+  Linux  — ~/.claude/.credentials.json
+  macOS  — macOS Keychain, service "Claude Code-credentials"
+            (file path is tried first; Keychain is the fallback)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import subprocess
+import sys
+from datetime import datetime, timezone
+from pathlib import Path
+
+from ...log import die
+
+
+_KEYCHAIN_SERVICE = "Claude Code-credentials"
+
+
+def claude_auth_path(host_env: dict[str, str] | None = None) -> Path:
+    env = os.environ if host_env is None else host_env
+    home = env.get("HOME")
+    if home:
+        return Path(home) / ".claude" / ".credentials.json"
+    return Path.home() / ".claude" / ".credentials.json"
+
+
+def _read_keychain() -> dict[str, object] | None:
+    """Try the macOS Keychain. Returns parsed JSON dict or None."""
+    if sys.platform != "darwin":
+        return None
+    try:
+        result = subprocess.run(
+            ["security", "find-generic-password", "-s", _KEYCHAIN_SERVICE, "-w"],
+            capture_output=True,
+            text=True,
+            timeout=10,
+        )
+    except (FileNotFoundError, subprocess.TimeoutExpired):
+        return None
+    if result.returncode != 0 or not result.stdout.strip():
+        return None
+    try:
+        raw = json.loads(result.stdout.strip())
+    except json.JSONDecodeError:
+        return None
+    return raw if isinstance(raw, dict) else None
+
+
+def claude_host_access_token(
+    host_env: dict[str, str] | None = None,
+    *,
+    now: datetime | None = None,
+) -> str:
+    path = claude_auth_path(host_env)
+    raw: dict[str, object] | None = None
+
+    if path.is_file():
+        try:
+            raw = json.loads(path.read_text())
+        except (OSError, json.JSONDecodeError) as e:
+            die(f"claude host credentials: could not read valid JSON at {path}: {e}")
+        if not isinstance(raw, dict):
+            die(f"claude host credentials: {path} must contain a JSON object")
+    else:
+        raw = _read_keychain()
+        if raw is None:
+            die(
+                f"claude host credentials: auth file missing at {path} and "
+                f"macOS Keychain lookup for '{_KEYCHAIN_SERVICE}' failed. "
+                "Run `claude login` on the host or disable "
+                "agent_provider.forward_host_credentials."
+            )
+
+    oauth = raw.get("claudeAiOauth")
+    if not isinstance(oauth, dict):
+        die(
+            "claude host credentials: claudeAiOauth is missing from credentials. "
+            "Run `claude login` on the host or disable "
+            "agent_provider.forward_host_credentials."
+        )
+
+    access_token = oauth.get("accessToken")
+    if not isinstance(access_token, str) or not access_token:
+        die(
+            "claude host credentials: claudeAiOauth.accessToken is missing or empty. "
+            "Run `claude login` on the host and restart the bottle."
+        )
+
+    # expiresAt is in milliseconds
+    expires_at = oauth.get("expiresAt")
+    if isinstance(expires_at, (int, float)):
+        check_now = now or datetime.now(timezone.utc)
+        exp_dt = datetime.fromtimestamp(float(expires_at) / 1000.0, timezone.utc)
+        if exp_dt <= check_now:
+            die(
+                "claude host credentials: host Claude access token is expired. "
+                "Run `claude login` on the host and restart the bottle."
+            )
+
+    return access_token
+
+
+__all__ = [
+    "claude_auth_path",
+    "claude_host_access_token",
+]

bot_bottle/contrib/codex/Dockerfile

@@ -3,10 +3,17 @@
 # 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-slim
+FROM node:22-trixie-slim
 
 RUN apt-get update \
-  && apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep \
+  && apt-get install -y --no-install-recommends \
+  git \
+  ca-certificates \
+  curl \
+  openssh-client \
+  podman \
+  procps \
+  ripgrep \
   && rm -rf /var/lib/apt/lists/*
 
 # App-specific deps. Python isn't required by codex itself
@@ -17,6 +24,8 @@ RUN apt-get update \
   && apt-get install -y --no-install-recommends python3 python3-pip python3-venv \
   && rm -rf /var/lib/apt/lists/*
 
+RUN install -d -o node -g node -m 755 /home/node/.config /home/node/.config/git
+
 USER node
 WORKDIR /home/node
 

bot_bottle/contrib/pi/Dockerfile

@@ -2,7 +2,7 @@
 #
 # Node LTS, git/network tooling, and the Pi coding-agent CLI installed globally.
 
-FROM node:22-slim
+FROM node:22-trixie-slim
 
 RUN apt-get update \
   && apt-get install -y --no-install-recommends \
@@ -10,6 +10,8 @@ RUN apt-get update \
   ca-certificates \
   curl \
   fd-find \
+  openssh-client \
+  podman \
   ripgrep \
   && ln -s /usr/bin/fdfind /usr/local/bin/fd \
   && rm -rf /var/lib/apt/lists/*
@@ -21,7 +23,8 @@ RUN apt-get update \
 RUN npm install -g --ignore-scripts --no-fund --no-audit @earendil-works/pi-coding-agent \
   && npm cache clean --force
 
-RUN mkdir -p /home/node/.pi/agent \
+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 \
   /tmp/pi-subagents-uid-1000 \
   && chown -R node:node /home/node/.pi /tmp \

bot_bottle/egress.py

@@ -30,6 +30,7 @@ if TYPE_CHECKING:
     from .manifest import ManifestBottle
 
 CODEX_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CODEX_HOST_ACCESS_TOKEN"
+CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"
 
 EGRESS_HOSTNAME = "egress"
 
@@ -145,6 +146,7 @@ def egress_manifest_routes(
             outbound_detectors=r.OutboundDetectors,
             inbound_detectors=r.InboundDetectors,
             outbound_on_match=r.OutboundOnMatch,
+            preserve_auth=r.PreserveAuth,
         ))
     return tuple(out)
 
@@ -400,6 +402,7 @@ class Egress(ABC):
         )
 
 __all__ = [
+    "CLAUDE_HOST_CREDENTIAL_TOKEN_REF",
     "CODEX_HOST_CREDENTIAL_TOKEN_REF",
     "EGRESS_HOSTNAME",
     "EGRESS_ROUTES_FILENAME",

bot_bottle/egress_addon.py

@@ -367,7 +367,10 @@ class EgressAddon:
 
         # Strip agent-set Authorization after DLP scan so smuggled tokens
         # are caught above; the route may inject gateway-owned auth below.
-        flow.request.headers.pop("authorization", None)
+        # 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:
+            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()}

bot_bottle/egress_addon_core.py

@@ -78,6 +78,7 @@ class Route:
     inbound_detectors: tuple[str, ...] | None = None
     # "" means unset → DEFAULT_OUTBOUND_ON_MATCH. See OUTBOUND_ON_MATCH_VALUES.
     outbound_on_match: str = ""
+    preserve_auth: bool = False
 
 
 LOG_OFF = 0    # no logging
@@ -308,11 +309,18 @@ def _parse_one(idx: int, raw: object) -> Route:
         idx, host, raw_dict,
     )
 
+    preserve_auth_raw = raw_dict.get("preserve_auth", False)
+    if preserve_auth_raw is not True and preserve_auth_raw is not False:
+        raise ValueError(
+            f"{label} ({host}): 'preserve_auth' must be a boolean"
+        )
+    preserve_auth: bool = preserve_auth_raw
+
     for k in raw_dict:
-        if k not in ("host", "matches", "auth_scheme", "token_env", "dlp", "git"):
+        if k not in ("host", "matches", "auth_scheme", "token_env", "dlp", "git", "preserve_auth"):
             raise ValueError(
                 f"{label} ({host}): unknown key {k!r}; accepted keys "
-                f"are 'host', 'matches', 'auth_scheme', 'token_env', 'dlp', 'git'"
+                f"are 'host', 'matches', 'auth_scheme', 'token_env', 'dlp', 'git', 'preserve_auth'"
             )
 
     return Route(
@@ -324,6 +332,7 @@ def _parse_one(idx: int, raw: object) -> Route:
         outbound_detectors=outbound_detectors,
         inbound_detectors=inbound_detectors,
         outbound_on_match=outbound_on_match,
+        preserve_auth=preserve_auth,
     )
 
 
@@ -376,6 +385,8 @@ def route_to_yaml_dict(r: Route) -> dict[str, object]:
         dlp["outbound_on_match"] = r.outbound_on_match
     if dlp:
         d["dlp"] = dlp
+    if r.preserve_auth:
+        d["preserve_auth"] = True
     return d
 
 

bot_bottle/gateway_init.py

@@ -61,6 +61,11 @@ class _DaemonSpec:
 _EGRESS_ONLY_ENV_PREFIXES: tuple[str, ...] = ("EGRESS_TOKEN_",)
 _READY_GATED_DAEMONS: tuple[str, ...] = ("git-gate", "git-http")
 
+# Daemons that must be requested explicitly via BOT_BOTTLE_GATEWAY_DAEMONS
+# and are NOT started in the default (env-var-unset) case. The orchestrator
+# only runs in the combined infra container, never in a standalone gateway.
+_OPT_IN_DAEMONS: frozenset[str] = frozenset({"orchestrator"})
+
 
 def _env_for_daemon(name: str, base_env: dict[str, str]) -> dict[str, str]:
     """Egress sees the full bundle env. Everyone else gets a copy
@@ -75,7 +80,14 @@ def _env_for_daemon(name: str, base_env: dict[str, str]) -> dict[str, str]:
     }
 
 
+# The orchestrator is listed first so it starts before the gateway daemons,
+# giving the control plane a head start to accept /resolve calls. The gateway
+# daemons tolerate early /resolve failures and retry per-request.
 _DAEMONS: tuple[_DaemonSpec, ...] = (
+    _DaemonSpec("orchestrator", (
+        "python3", "-m", "bot_bottle.orchestrator",
+        "--host", "0.0.0.0", "--port", "8099", "--broker", "stub",
+    )),
     _DaemonSpec("egress", ("/bin/sh", "/app/egress-entrypoint.sh")),
     _DaemonSpec("git-gate", ("/bin/sh", "/git-gate-entrypoint.sh")),
     _DaemonSpec("git-http", ("python3", "-m", "bot_bottle.git_http_backend")),
@@ -103,18 +115,20 @@ def _selected_daemons(
     env: dict[str, str],
     all_daemons: Sequence[_DaemonSpec] | None = None,
 ) -> tuple[_DaemonSpec, ...]:
-    """Filter the daemon set by the BOT_BOTTLE_GATEWAY_DAEMONS env
-    var. Unknown names in the list are ignored — the caller is the
-    source of truth for which daemons are wired.
+    """Filter the daemon set by the BOT_BOTTLE_GATEWAY_DAEMONS env var.
 
-    `all_daemons` defaults to `_DAEMONS` resolved at call time (not
-    at definition time), so tests can monkey-patch the module-level
-    `_DAEMONS` and have the new value take effect."""
+    When the var is unset/empty, return all non-opt-in daemons (the
+    standard gateway subset). Opt-in daemons (e.g. `orchestrator`) only
+    run when explicitly named — they never start in a plain gateway
+    container that doesn't set the env var. Unknown names are ignored.
+
+    `all_daemons` defaults to `_DAEMONS` resolved at call time (not at
+    definition time), so tests can pass a custom list."""
     if all_daemons is None:
         all_daemons = _DAEMONS
     raw = env.get("BOT_BOTTLE_GATEWAY_DAEMONS", "").strip()
     if not raw:
-        return tuple(all_daemons)
+        return tuple(d for d in all_daemons if d.name not in _OPT_IN_DAEMONS)
     wanted = {n.strip() for n in raw.split(",") if n.strip()}
     return tuple(d for d in all_daemons if d.name in wanted)
 
@@ -136,7 +150,7 @@ def _pump(name: str, stream: IO[bytes]) -> None:
 
 def _spawn(spec: _DaemonSpec) -> subprocess.Popen[bytes]:
     env = _env_for_daemon(spec.name, dict(os.environ))
-    proc = subprocess.Popen(
+    proc = subprocess.Popen(  # pylint: disable=consider-using-with
         _argv_for_daemon(spec.name, spec.argv, env),
         stdout=subprocess.PIPE,
         stderr=subprocess.STDOUT,
@@ -183,6 +197,14 @@ class _Supervisor:
                 except ProcessLookupError:
                     pass
 
+    def _sigkill_all(self) -> None:
+        for _, p in self.procs:
+            if p.poll() is None:
+                try:
+                    p.kill()
+                except ProcessLookupError:
+                    pass
+
     def request_restart(self, daemon_name: str) -> bool:
         """Queue a daemon restart for the main loop to process.
 
@@ -235,12 +257,7 @@ class _Supervisor:
                         f"grace ({_GRACE_SECONDS:.0f}s) elapsed; SIGKILL on "
                         f"{', '.join(still_running)}"
                     )
-                    for _, p in self.procs:
-                        if p.poll() is None:
-                            try:
-                                p.kill()
-                            except ProcessLookupError:
-                                pass
+                    self._sigkill_all()
 
         done = all(p.poll() is not None for _, p in self.procs)
         if done:
@@ -361,7 +378,10 @@ def main(argv: Sequence[str] | None = None) -> int:
     # --signal HUP <bundle>` after writing routes.yaml. The kernel
     # delivers SIGHUP to PID 1 (this supervisor); forward it to
     # mitmdump so it reloads its addon.
-    signal.signal(signal.SIGHUP, lambda *_: sup.forward_signal(signal.SIGHUP, "egress"))  # type: ignore
+    signal.signal(
+        signal.SIGHUP,
+        lambda *_: sup.forward_signal(signal.SIGHUP, "egress"),  # type: ignore[misc]
+    )
 
     while not sup.tick():
         time.sleep(_POLL_INTERVAL)

bot_bottle/image_cache.py

@@ -0,0 +1,42 @@
+"""Shared helpers for cached-image quickstart stale checks."""
+
+from __future__ import annotations
+
+from datetime import datetime, timezone
+from pathlib import Path
+
+try:
+    from .config_store import ConfigStore
+except ImportError:
+    from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+
+
+class StaleImageError(Exception):
+    """Raised when a cached image or artifact exceeds the configured staleness
+    threshold. Callers can catch this to prompt interactively; headless paths
+    let it propagate as a fatal error."""
+
+
+def check_stale(label: str, created_at: datetime) -> None:
+    """Raise StaleImageError if `created_at` is older than the configured
+    stale-warning threshold. Negative threshold disables the check."""
+    threshold_days = ConfigStore().cached_image_stale_warning_days()
+    if threshold_days < 0:
+        return
+    now = datetime.now(timezone.utc)
+    created = created_at.astimezone(timezone.utc)
+    age = now - created
+    if age.total_seconds() <= threshold_days * 86400:
+        return
+    raise StaleImageError(
+        f"cached {label} is {age.days} day(s) old; "
+        "quickstart does not verify it matches the current Dockerfile/context"
+    )
+
+
+def check_stale_path(label: str, path: Path) -> None:
+    """Raise StaleImageError if `path`'s mtime exceeds the staleness threshold."""
+    check_stale(label, datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc))
+
+
+__all__ = ["StaleImageError", "check_stale", "check_stale_path"]

bot_bottle/manifest_agent.py

@@ -25,8 +25,9 @@ class ManifestAgentProvider:
     header, and sets a placeholder CLAUDE_CODE_OAUTH_TOKEN in the agent
     so the Claude Code CLI starts.
 
-    `forward_host_credentials` forwards the host Codex auth token into
-    the egress daemon (Codex only).
+    `forward_host_credentials` forwards the host provider auth token into
+    the egress sidecar (Codex and Claude). For Codex this reads
+    `~/.codex/auth.json`; for Claude it reads `~/.claude/.credentials.json`.
     """
 
     template: str = "claude"
@@ -92,10 +93,15 @@ class ManifestAgentProvider:
                 f"is only supported for built-in templates "
                 f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
             )
-        if forward_host_credentials and template != "codex":
+        if forward_host_credentials and template not in {"codex", "claude"}:
             raise ManifestError(
                 f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
-                "is currently only supported for template 'codex'"
+                "is only supported for templates 'codex' and 'claude'"
+            )
+        if forward_host_credentials and auth_token:
+            raise ManifestError(
+                f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
+                "and auth_token both set; use one or the other"
             )
         settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
         return cls(

bot_bottle/manifest_bottle.py

@@ -44,9 +44,6 @@ class ManifestBottle:
     # daemon that exposes egress MCP tools to the agent. Set
     # `supervise: false` to skip the gateway.
     supervise: bool = True
-    # Experimental guest-local container engine (issue #392). Backends must
-    # implement this without granting access to a host/shared daemon.
-    docker_access: bool = False
 
     @classmethod
     def from_dict(cls, name: str, raw: object) -> "ManifestBottle":
@@ -126,15 +123,7 @@ class ManifestBottle:
                 f"(was {type(supervise_raw).__name__})"
             )
 
-        docker_access_raw = d.get("docker_access", False)
-        if not isinstance(docker_access_raw, bool):
-            raise ManifestError(
-                f"bottle '{name}' docker_access must be a boolean "
-                f"(was {type(docker_access_raw).__name__})"
-            )
-
         return cls(
             env=env, agent_provider=agent_provider, git=git,
             git_user=git_user, egress=egress, supervise=supervise_raw,
-            docker_access=docker_access_raw,
         )

bot_bottle/manifest_egress.py

@@ -71,6 +71,7 @@ class ManifestEgressRoute:
     OutboundDetectors: tuple[str, ...] | None = None
     InboundDetectors: tuple[str, ...] | None = None
     OutboundOnMatch: str = ""
+    PreserveAuth: bool = False
 
     @classmethod
     def from_dict(cls, bottle_name: str, idx: int, raw: object) -> "ManifestEgressRoute":
@@ -190,11 +191,22 @@ class ManifestEgressRoute:
                         f"only 'fetch' is accepted"
                     )
 
+        # --- preserve_auth ---
+        preserve_auth = False
+        if "preserve_auth" in d:
+            raw_preserve_auth = d.get("preserve_auth")
+            if not isinstance(raw_preserve_auth, bool):
+                raise ManifestError(
+                    f"{label} preserve_auth must be a boolean "
+                    f"(was {type(raw_preserve_auth).__name__})"
+                )
+            preserve_auth = raw_preserve_auth
+
         for k in d:
-            if k not in ("host", "matches", "auth", "role", "dlp", "git"):
+            if k not in ("host", "matches", "auth", "role", "dlp", "git", "preserve_auth"):
                 raise ManifestError(
                     f"{label} has unknown key {k!r}; accepted keys are "
-                    f"'host', 'matches', 'auth', 'role', 'dlp', 'git'"
+                    f"'host', 'matches', 'auth', 'role', 'dlp', 'git', 'preserve_auth'"
                 )
 
         return cls(
@@ -207,6 +219,7 @@ class ManifestEgressRoute:
             OutboundDetectors=outbound_detectors,
             InboundDetectors=inbound_detectors,
             OutboundOnMatch=outbound_on_match,
+            PreserveAuth=preserve_auth,
         )
 
 

bot_bottle/manifest_extends.py

@@ -54,7 +54,6 @@ def _merge_two_bottles_runtime(base: "ManifestBottle", override: "ManifestBottle
         git_user=merged_git_user,
         egress=merged_egress,
         supervise=override.supervise,
-        docker_access=override.docker_access,
     )
 
 
@@ -207,7 +206,6 @@ def _fold_two_bottles(
         git_user=merged_git_user,
         egress=merged_egress,
         supervise=later.supervise,
-        docker_access=later.docker_access,
     ), merged_repos_raw
 
 
@@ -268,11 +266,6 @@ def _merge_bottles(
     merged_supervise = (
         child.supervise if "supervise" in child_raw else parent.supervise
     )
-    merged_docker_access = (
-        child.docker_access
-        if "docker_access" in child_raw
-        else parent.docker_access
-    )
     validate_egress_routes(name, merged_egress.routes)
 
     return ManifestBottle(
@@ -282,7 +275,6 @@ def _merge_bottles(
         git_user=merged_git_user,
         egress=merged_egress,
         supervise=merged_supervise,
-        docker_access=merged_docker_access,
     )
 
 

bot_bottle/manifest_schema.py

@@ -16,10 +16,7 @@ _FILENAME_RX = re.compile(r"^[a-z][a-z0-9-]*$")
 # sets dies with a "did you mean" pointer: typos should not silently
 # ghost into an empty config.
 BOTTLE_KEYS = frozenset(
-    {
-        "env", "extends", "agent_provider", "git-gate", "egress", "supervise",
-        "docker_access",
-    }
+    {"env", "extends", "agent_provider", "git-gate", "egress", "supervise"}
 )
 AGENT_KEYS_REQUIRED: frozenset[str] = frozenset()
 AGENT_KEYS_OPTIONAL = frozenset({"bottle", "skills", "git-gate"})

bot_bottle/orchestrator/client.py

@@ -41,10 +41,13 @@ class OrchestratorClientError(RuntimeError):
 @dataclass(frozen=True)
 class RegisteredBottle:
     """What `POST /bottles` returns: the minted bottle id and the per-bottle
-    identity token the agent presents for app-layer attribution."""
+    identity token the agent presents for app-layer attribution. `env_var_secret`
+    is set by the caller (not from the server response) and carries the
+    encryption key so it can be injected into the agent container's env."""
 
     bottle_id: str
     identity_token: str
+    env_var_secret: str = ""
 
 
 class OrchestratorClient:
@@ -120,17 +123,21 @@ class OrchestratorClient:
         metadata: str = "",
         policy: str = "",
         tokens: dict[str, str] | None = None,
+        env_var_secret: str = "",
     ) -> RegisteredBottle:
         """Register a bottle and broker its launch (`POST /bottles`). `tokens`
         are the per-bottle egress auth values (env_name -> value) the
-        orchestrator holds in memory for the gateway to inject. Returns the
-        minted id + identity token."""
+        orchestrator holds in memory for the gateway to inject. When
+        *env_var_secret* is provided, the orchestrator also encrypts the token
+        values and stores them in ``bottled_agent_secrets`` for restart
+        recovery. Returns the minted id + identity token."""
         payload = self._ok("POST", "/bottles", {
             "source_ip": source_ip,
             "image_ref": image_ref,
             "metadata": metadata,
             "policy": policy,
             "tokens": tokens or {},
+            "env_var_secret": env_var_secret,
         })
         bottle_id = payload.get("bottle_id")
         token = payload.get("identity_token")
@@ -138,6 +145,24 @@ class OrchestratorClient:
             raise OrchestratorClientError("register: response missing bottle_id/identity_token")
         return RegisteredBottle(bottle_id=bottle_id, identity_token=token)
 
+    def reprovision_gateway(self, bottle_id: str, env_var_secret: str) -> bool:
+        """Re-inject a bottle's egress tokens from its ENV_VAR_SECRET
+        (`POST /bottles/<id>/reprovision_gateway`). Returns True when the
+        orchestrator successfully decrypted and restored the tokens, False
+        when it had no stored secrets for this bottle (404)."""
+        status, _ = self._request(
+            "POST",
+            f"/bottles/{bottle_id}/reprovision_gateway",
+            {"env_var_secret": env_var_secret},
+        )
+        if status == 404:
+            return False
+        if not 200 <= status < 300:
+            raise OrchestratorClientError(
+                f"reprovision_gateway {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."""

bot_bottle/orchestrator/control_plane.py

@@ -9,9 +9,13 @@ vsock / unix-socket portability caveats):
     GET    /bottles                   -> 200 {"bottles": [ <redacted record>, ...]}
     POST   /bottles                   -> 201 {"bottle_id","identity_token"}  (launch)
                                          body: {"source_ip", ["image_ref"],
-                                                ["metadata"], ["policy"]}
+                                                ["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": [...],
@@ -116,12 +120,14 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
         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}
 
@@ -138,6 +144,23 @@ def dispatch(  # pylint: disable=too-many-return-statements,too-many-branches
             return 200, {"updated": True}
         return 404, {"error": "no such bottle"}
 
+    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"}
+
     if method == "DELETE" and route.startswith("/bottles/"):
         bottle_id = route[len("/bottles/"):]
         if orch.teardown_bottle(bottle_id):

bot_bottle/orchestrator/gateway.py

@@ -27,6 +27,7 @@ from ..paths import (
     CONTROL_PLANE_TOKEN_ENV,
     host_control_plane_token,
     host_db_path,
+    host_gateway_ca_dir,
 )
 from ..supervise import DB_PATH_IN_CONTAINER
 
@@ -48,14 +49,23 @@ GATEWAY_LABEL = "bot-bottle-orch-gateway=1"
 # the source IP the gateway attributes by is the address on this network.
 GATEWAY_NETWORK = "bot-bottle-gateway"
 
-# mitmproxy's CA dir in the bundle. A persistent named volume here keeps the
-# gateway's self-generated CA STABLE across container recreation — every agent
-# installs this one CA to trust the shared gateway's TLS interception, so it
-# must not rotate when the gateway restarts.
+# mitmproxy's CA dir in the bundle. The host's gateway-CA dir (see
+# `host_gateway_ca_dir`) is bind-mounted here so the gateway's self-generated
+# CA stays STABLE across container recreation — every agent installs this one
+# CA to trust the shared gateway's TLS interception, so it must not rotate when
+# the gateway restarts. A host bind-mount rather than a named volume: a named
+# volume is silently wiped by `docker volume prune`, minting a fresh CA that
+# breaks every running bottle (issue #450).
 MITMPROXY_HOME = "/home/mitmproxy/.mitmproxy"
-GATEWAY_CA_VOLUME = "bot-bottle-gateway-mitmproxy"
 GATEWAY_CA_CERT = f"{MITMPROXY_HOME}/mitmproxy-ca-cert.pem"
 
+# The CA material mitmproxy writes into its confdir. mitmproxy reuses these on
+# startup when present and generates them only on first run, so persisting them
+# is what makes the CA stable; deleting them (see `rotate_gateway_ca`) forces a
+# fresh CA on the next start. `mitmproxy-ca.pem` (cert + private key) is the
+# signing identity; the rest are derived encodings agents/clients consume.
+GATEWAY_CA_GLOB = "mitmproxy-ca*"
+
 # The gateway data-plane image + its Dockerfile. Kept as a local constant
 # rather than imported from the backend layer, which would drag
 # the whole backend layer into the lean orchestrator (see #359); unify when
@@ -73,6 +83,26 @@ def _host_db_dir() -> str:
     return str(db_dir)
 
 
+def rotate_gateway_ca(ca_dir: Path | None = None) -> list[Path]:
+    """Delete the persisted mitmproxy CA so the next gateway start mints a
+    fresh one — the explicit, deliberate CA-rollover path (issue #450).
+
+    Persistence keeps the CA stable across restarts precisely because mitmproxy
+    reuses the on-disk CA; rotation is therefore just removing that material.
+    Returns the files removed (empty when there was no CA yet); idempotent.
+
+    This only clears the on-disk CA. It does NOT stop the running gateway (whose
+    mitmproxy still holds the old CA in memory) or re-provision agents — the
+    caller recreates the gateway to mint the new CA and re-attaches bottles.
+    `rotate-ca` on the orchestrator CLI wires those steps together."""
+    ca_dir = ca_dir if ca_dir is not None else host_gateway_ca_dir()
+    removed: list[Path] = []
+    for path in sorted(ca_dir.glob(GATEWAY_CA_GLOB)):
+        path.unlink()
+        removed.append(path)
+    return removed
+
+
 class GatewayError(Exception):
     """The shared gateway failed to build/start/stop (non-zero `docker` exit)."""
 
@@ -221,9 +251,10 @@ class DockerGateway(Gateway):
             "--name", self.name,
             "--label", GATEWAY_LABEL,
             "--network", self.network,
-            # Persist the self-generated CA so it survives restarts (agents
-            # trust it) — see GATEWAY_CA_VOLUME.
-            "--volume", f"{GATEWAY_CA_VOLUME}:{MITMPROXY_HOME}",
+            # Persist the self-generated CA on the host so it survives both
+            # container recreation AND docker volume pruning (agents trust it)
+            # — see host_gateway_ca_dir / issue #450.
+            "--volume", f"{host_gateway_ca_dir()}:{MITMPROXY_HOME}",
             # Share the one host DB: the supervise daemon queues proposals
             # into the same file the orchestrator (and the operator, over
             # HTTP) reads — no second, disconnected DB in the container.
@@ -272,7 +303,7 @@ class DockerGateway(Gateway):
 
 
 __all__ = [
-    "Gateway", "DockerGateway", "GatewayError",
+    "Gateway", "DockerGateway", "GatewayError", "rotate_gateway_ca",
     "GATEWAY_NAME", "GATEWAY_LABEL", "GATEWAY_IMAGE", "GATEWAY_NETWORK",
-    "GATEWAY_CA_VOLUME", "GATEWAY_CA_CERT",
+    "GATEWAY_CA_CERT", "GATEWAY_CA_GLOB",
 ]

bot_bottle/orchestrator/lifecycle.py

@@ -1,17 +1,15 @@
 """Orchestrator + gateway lifecycle (PRD 0070, docker slice).
 
-Runs the orchestrator control plane **as a container** on the shared gateway
-network, alongside the gateway container. This is the PRD's "virtualize the
-orchestrator": container↔container between the gateway and the orchestrator
-avoids the host firewall (which drops container→host traffic), and the gateway
-reaches the control plane by container name over docker DNS. The host CLI
-reaches it via a published loopback port.
+Runs both the orchestrator control plane and the gateway data plane inside
+a single `bot-bottle-infra` container on the shared gateway network —
+matching the structure already used by the macOS and Firecracker backends.
+`gateway_init` is PID 1 and supervises both; the infra container is an
+idempotent per-host singleton.
 
-The orchestrator runs with the **register-only broker** — the *backend*
-launches agent containers (compose), so the orchestrator needs no docker
-socket. That keeps this control-plane container unprivileged; the host manages
-both containers. `ensure_running` is an idempotent singleton (fixed container
-names + the published port).
+The combined container replaces the prior two-container split
+(bot-bottle-orchestrator + bot-bottle-orch-gateway). The host CLI reaches
+the control plane via a published loopback port; gateway daemons reach it
+over 127.0.0.1 (same container).
 """
 
 from __future__ import annotations
@@ -25,50 +23,75 @@ from pathlib import Path
 
 from .. import log
 from ..docker_cmd import run_docker
-from ..paths import CONTROL_PLANE_TOKEN_ENV, bot_bottle_root, host_control_plane_token
-from .gateway import GATEWAY_IMAGE, GATEWAY_NAME, GATEWAY_NETWORK, DockerGateway, GatewayError
+from ..paths import (
+    CONTROL_PLANE_TOKEN_ENV,
+    bot_bottle_root,
+    host_control_plane_token,
+    host_gateway_ca_dir,
+)
+from ..supervise import DB_PATH_IN_CONTAINER
+from .gateway import (
+    GATEWAY_DOCKERFILE,
+    GATEWAY_IMAGE,
+    GATEWAY_NETWORK,
+    GatewayError,
+    MITMPROXY_HOME,
+    _host_db_dir,
+)
 
 DEFAULT_PORT = 8099
-ORCHESTRATOR_NAME = "bot-bottle-orchestrator"
-ORCHESTRATOR_LABEL = "bot-bottle-orchestrator=1"
-# The control-plane's own runtime image — lean (python + the stdlib-only
-# `bot_bottle` package, bind-mounted at run time), distinct from the heavy
-# gateway data-plane image it used to borrow (#384). Env override for
-# operators pinning a published build.
+DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
+
+INFRA_NAME = "bot-bottle-infra"
+INFRA_LABEL = "bot-bottle-infra=1"
+# The combined infra image: gateway data plane + orchestrator content.
+# Built from Dockerfile.infra (FROM gateway + COPY --from orchestrator).
+INFRA_IMAGE = os.environ.get("BOT_BOTTLE_INFRA_IMAGE", "bot-bottle-infra:latest")
+INFRA_DOCKERFILE = "Dockerfile.infra"
+# Baked as a container label so `ensure_running` can detect whether the
+# running container is executing the current bind-mounted source.
+INFRA_SOURCE_HASH_LABEL = "bot-bottle-infra-source-hash"
+
+# Orchestrator image: the single canonical definition of the control-plane
+# content (lean: python:3.12-slim + bot_bottle package, no mitmproxy/git).
+# Used as a build intermediate: `Dockerfile.infra` COPY --from this image.
 ORCHESTRATOR_IMAGE = os.environ.get(
     "BOT_BOTTLE_ORCHESTRATOR_IMAGE", "bot-bottle-orchestrator:latest"
 )
 ORCHESTRATOR_DOCKERFILE = "Dockerfile.orchestrator"
-# Baked onto the container as a label so `ensure_running` can tell whether the
-# running process is executing the *current* bind-mounted source — see
-# `source_hash`.
-ORCHESTRATOR_SOURCE_HASH_LABEL = "bot-bottle-orchestrator-source-hash"
 
-# The repo root is bind-mounted into the control-plane container so
-# `python -m bot_bottle.orchestrator` resolves the package (the orchestrator
-# is stdlib-only, so the lean orchestrator image's python is enough).
-_REPO_ROOT = Path(__file__).resolve().parents[2]
-_APP_DIR = "/app"
+# The gateway daemons + orchestrator the infra container runs.
+# BOT_BOTTLE_GATEWAY_DAEMONS listing `orchestrator` opts it in to
+# gateway_init's supervise tree (see gateway_init._OPT_IN_DAEMONS).
+_INFRA_DAEMONS = "egress,git-http,supervise,orchestrator"
+
+# The bind-mount path for the live control-plane source inside the
+# container. Separate from /app so the gateway's baked scripts
+# (egress_addon.py, egress-entrypoint.sh) are not overlaid.
+_SRC_IN_CONTAINER = "/bot-bottle-src"
+# Bot-bottle host-root bind-mount inside the container (DB + state).
 _ROOT_IN_CONTAINER = "/bot-bottle-root"
 
+# The supervise daemon writes proposals into the host DB directory.
+_SUPERVISE_DB_DIR_IN_CONTAINER = os.path.dirname(DB_PATH_IN_CONTAINER)
+
 _HEALTH_POLL_SECONDS = 0.25
-DEFAULT_STARTUP_TIMEOUT_SECONDS = 45.0
 _HEALTH_REQUEST_TIMEOUT_SECONDS = 1.0
 
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
 
 class OrchestratorStartError(RuntimeError):
-    """The orchestrator container did not become healthy within the timeout."""
+    """The infra container did not become healthy within the timeout."""
 
 
 def source_hash(repo_root: Path) -> str:
     """Content hash of the orchestrator's bind-mounted Python source (the
-    `bot_bottle` package the control-plane process imports). This only
-    changes when the code that would actually run inside the container
-    changes — `ensure_running` recreates the container on a mismatch and
-    otherwise leaves a healthy one alone, so a bottle launch that isn't
-    accompanied by a code change doesn't restart the process and drop every
-    *other* active bottle's in-memory egress tokens (`Orchestrator._tokens`
-    in `service.py`, never persisted to disk by design)."""
+    `bot_bottle` package the control-plane process imports). Changes only
+    when the code that would actually run changes — `ensure_running`
+    recreates the container on a mismatch so a code change takes effect,
+    but leaves a healthy up-to-date container alone to preserve in-memory
+    egress tokens."""
     h = hashlib.sha256()
     for path in sorted((repo_root / "bot_bottle").rglob("*.py")):
         h.update(str(path.relative_to(repo_root)).encode())
@@ -77,57 +100,37 @@ def source_hash(repo_root: Path) -> str:
 
 
 class OrchestratorService:
-    """Manages the orchestrator control-plane container + the shared gateway.
+    """Manages the single per-host infra container (control plane + gateway).
     Callers only need `ensure_running()` + `url`.
 
-    `orchestrator_name` / `orchestrator_label` let backends run independent
-    orchestrators on the same host without name collisions (e.g. the
-    Firecracker backend uses `bot-bottle-fc-orchestrator` alongside the Docker
-    backend's `bot-bottle-orchestrator`); `gateway_name` gives the paired
-    gateway container the same treatment (e.g. isolated integration tests
-    that can't share the production `GATEWAY_NAME` singleton). Subclass and
-    override `_gateway()` for anything `_gateway_image`/`gateway_name` can't
-    express (a genuinely backend-specific gateway variant)."""
+    `infra_name` / `infra_label` let backends run independent infra containers
+    on the same host without name collisions (e.g. isolated integration tests
+    that can't share the production INFRA_NAME singleton)."""
 
     def __init__(
         self,
         *,
         port: int = DEFAULT_PORT,
         network: str = GATEWAY_NETWORK,
-        image: str = ORCHESTRATOR_IMAGE,
-        gateway_image: str = GATEWAY_IMAGE,
-        gateway_name: str = GATEWAY_NAME,
+        image: str = INFRA_IMAGE,
         repo_root: Path = _REPO_ROOT,
         host_root: Path | None = None,
-        orchestrator_name: str = ORCHESTRATOR_NAME,
-        orchestrator_label: str = ORCHESTRATOR_LABEL,
+        infra_name: str = INFRA_NAME,
+        infra_label: str = INFRA_LABEL,
     ) -> None:
         self.port = port
         self.network = network
-        # Two distinct images (#384): `image` is the lean control-plane
-        # runtime this container runs; `_gateway_image` is the heavy egress /
-        # git-gate / supervise data plane the gateway container runs. They
-        # were one conflated image before the split.
         self.image = image
-        self._gateway_image = gateway_image
-        self._gateway_name = gateway_name
         self._repo_root = repo_root
         self._host_root = host_root or bot_bottle_root()
-        self._orchestrator_name = orchestrator_name
-        self._orchestrator_label = orchestrator_label
+        self._infra_name = infra_name
+        self._infra_label = infra_label
 
     @property
     def url(self) -> str:
         """Host-side control-plane URL (published loopback port)."""
         return f"http://127.0.0.1:{self.port}"
 
-    @property
-    def internal_url(self) -> str:
-        """Control-plane URL as the gateway container reaches it — by name over
-        docker DNS on the shared network. This is the gateway's
-        BOT_BOTTLE_ORCHESTRATOR_URL."""
-        return f"http://{self._orchestrator_name}:{self.port}"
-
     def is_healthy(self, *, timeout: float = _HEALTH_REQUEST_TIMEOUT_SECONDS) -> bool:
         try:
             with urllib.request.urlopen(f"{self.url}/health", timeout=timeout) as resp:
@@ -139,139 +142,131 @@ class OrchestratorService:
         proc = run_docker(["docker", "ps", "--filter", f"name=^/{name}$", "--format", "{{.Names}}"])
         return name in proc.stdout.split()
 
-    def _run_orchestrator_container(self, current_hash: str) -> None:
-        """Start the control-plane container (idempotent: clears a stale
-        fixed-name container first). Register-only broker → no docker socket.
-        Labels the container with `current_hash` so a later `ensure_running`
-        can detect a real code change (see `source_hash`)."""
-        run_docker(["docker", "rm", "--force", self._orchestrator_name])
-        proc = run_docker([
-            "docker", "run", "--detach",
-            "--name", self._orchestrator_name,
-            "--label", self._orchestrator_label,
-            "--label", f"{ORCHESTRATOR_SOURCE_HASH_LABEL}={current_hash}",
-            "--network", self.network,
-            # Host CLI reaches the control plane here; bound to loopback so it
-            # is not exposed on the host's external interfaces. NOTE: the
-            # container is still on `self.network` (the shared gateway network),
-            # so agents can reach it by container IP — which is exactly why the
-            # control plane requires the secret below rather than trusting the
-            # network boundary.
-            "--publish", f"127.0.0.1:{self.port}:{self.port}",
-            "--volume", f"{self._repo_root}:{_APP_DIR}:ro",
-            "--workdir", _APP_DIR,
-            # Persist the registry DB on the host (sole-owner: only the
-            # orchestrator opens bot-bottle.db).
-            "--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}",
-            "--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
-            # The control-plane secret it requires on every route but /health.
-            # Bare `--env NAME` → docker inherits the value from the run env
-            # below, so the secret never lands on argv / `docker inspect`.
-            "--env", CONTROL_PLANE_TOKEN_ENV,
-            "--entrypoint", "python3",
-            self.image,
-            "-m", "bot_bottle.orchestrator",
-            "--host", "0.0.0.0", "--port", str(self.port), "--broker", "stub",
-        ], env={**os.environ, CONTROL_PLANE_TOKEN_ENV: host_control_plane_token()})
-        if proc.returncode != 0:
-            raise OrchestratorStartError(
-                f"orchestrator container failed to start: {proc.stderr.strip()}"
-            )
-
-    def _gateway(self) -> DockerGateway:
-        return DockerGateway(
-            self._gateway_image,
-            name=self._gateway_name,
-            network=self.network,
-            orchestrator_url=self.internal_url,
-        )
-
-    def _ensure_orchestrator_image(self) -> None:
-        """Build the lean control-plane image from `Dockerfile.orchestrator`
-        when it's missing (#384). Cheap — a `FROM python:*-slim` base with no
-        deps to install, so the layer cache makes rebuilds a no-op. Unlike the
-        gateway image this is build-if-missing, not build-every-time: the
-        control plane bind-mounts its source, so a code change is caught by the
-        source-hash recreate (below), not by an image rebuild."""
-        if run_docker(["docker", "image", "inspect", self.image]).returncode == 0:
-            return
-        argv = ["docker", "build", "-t", self.image,
-                "-f", str(self._repo_root / ORCHESTRATOR_DOCKERFILE),
-                str(self._repo_root)]
-        if os.environ.get("BOT_BOTTLE_NO_CACHE"):
-            argv.insert(2, "--no-cache")
-        proc = run_docker(argv)
-        if proc.returncode != 0:
-            raise GatewayError(
-                f"orchestrator image build failed: {proc.stderr.strip()}"
-            )
-
-    def _orchestrator_source_current(self, current_hash: str) -> bool:
-        """True iff the running orchestrator container was created from the
-        *current* bind-mounted source. Mirrors `DockerGateway`'s
-        image-staleness check, but by content hash rather than image id since
-        the orchestrator runs bind-mounted source, not a built image."""
-        if not self._container_running(self._orchestrator_name):
+    def _infra_source_current(self, current_hash: str) -> bool:
+        """True iff the running infra container was started from the current
+        bind-mounted source. Mirrors the macOS backend's `_source_current`."""
+        if not self._container_running(self._infra_name):
             return False
         proc = run_docker([
             "docker", "inspect", "--format",
-            "{{ index .Config.Labels \"" + ORCHESTRATOR_SOURCE_HASH_LABEL + "\" }}",
-            self._orchestrator_name,
+            "{{ index .Config.Labels \"" + INFRA_SOURCE_HASH_LABEL + "\" }}",
+            self._infra_name,
         ])
         if proc.returncode != 0:
-            return True  # can't compare -> don't churn a working container
+            return True  # can't compare → don't churn a working container
         return proc.stdout.strip() == current_hash
 
+    def _ensure_network(self) -> None:
+        if run_docker(["docker", "network", "inspect", self.network]).returncode == 0:
+            return
+        proc = run_docker(["docker", "network", "create", self.network])
+        if proc.returncode != 0 and "already exists" not in proc.stderr:
+            raise GatewayError(
+                f"gateway network {self.network} failed to create: {proc.stderr.strip()}"
+            )
+
+    def _build_images(self) -> None:
+        """Build the gateway base, the orchestrator intermediate, then the
+        infra image. All are cache-aware: a no-op when nothing changed."""
+        for tag, dockerfile in (
+            (GATEWAY_IMAGE, GATEWAY_DOCKERFILE),
+            (ORCHESTRATOR_IMAGE, ORCHESTRATOR_DOCKERFILE),
+            (self.image, INFRA_DOCKERFILE),
+        ):
+            argv = ["docker", "build", "-t", tag,
+                    "-f", str(self._repo_root / dockerfile),
+                    str(self._repo_root)]
+            if os.environ.get("BOT_BOTTLE_NO_CACHE"):
+                argv.insert(2, "--no-cache")
+            proc = run_docker(argv)
+            if proc.returncode != 0:
+                raise GatewayError(f"{dockerfile} build failed: {proc.stderr.strip()}")
+
+    def _run_infra_container(self, current_hash: str) -> None:
+        """Start the combined infra container (idempotent: clears a stale
+        fixed-name container first). Labels the container with `current_hash`
+        so a later `ensure_running` can detect a real code change."""
+        self._ensure_network()
+        run_docker(["docker", "rm", "--force", self._infra_name])
+        proc = run_docker([
+            "docker", "run", "--detach",
+            "--name", self._infra_name,
+            "--label", self._infra_label,
+            "--label", f"{INFRA_SOURCE_HASH_LABEL}={current_hash}",
+            "--network", self.network,
+            # Host CLI reaches the control plane here (loopback only).
+            # gateway_init always starts the orchestrator on DEFAULT_PORT (8099)
+            # inside the container; self.port is the host-side published port.
+            "--publish", f"127.0.0.1:{self.port}:{DEFAULT_PORT}",
+            # Persist the mitmproxy CA on the host so it survives container
+            # recreation AND docker volume pruning (issue #450): every agent
+            # trusts this one CA, so a fresh one would break all running bottles.
+            "--volume", f"{host_gateway_ca_dir()}:{MITMPROXY_HOME}",
+            # Shared supervise DB (same file the operator reads over HTTP).
+            "--volume", f"{_host_db_dir()}:{_SUPERVISE_DB_DIR_IN_CONTAINER}",
+            "--env", f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
+            # Live control-plane source, mounted to a path that does not
+            # overlay the gateway's baked /app scripts.
+            "--volume", f"{self._repo_root}:{_SRC_IN_CONTAINER}:ro",
+            # PYTHONPATH lets the orchestrator (and other Python daemons)
+            # import the live source ahead of the installed package.
+            "--env", f"PYTHONPATH={_SRC_IN_CONTAINER}",
+            # Orchestrator registry DB on the host (sole writer: control plane).
+            "--volume", f"{self._host_root}:{_ROOT_IN_CONTAINER}",
+            "--env", f"BOT_BOTTLE_ROOT={_ROOT_IN_CONTAINER}",
+            # Control-plane secret: required by the orchestrator (to enforce)
+            # and by the gateway daemons (to present on /resolve calls).
+            "--env", CONTROL_PLANE_TOKEN_ENV,
+            # Gateway daemons reach the orchestrator over loopback at its
+            # fixed internal port (DEFAULT_PORT), independent of self.port.
+            "--env", f"BOT_BOTTLE_ORCHESTRATOR_URL=http://127.0.0.1:{DEFAULT_PORT}",
+            # Opt the orchestrator into gateway_init's supervise tree.
+            "--env", f"BOT_BOTTLE_GATEWAY_DAEMONS={_INFRA_DAEMONS}",
+            self.image,
+        ], env={**os.environ, CONTROL_PLANE_TOKEN_ENV: host_control_plane_token()})
+        if proc.returncode != 0:
+            raise OrchestratorStartError(
+                f"infra container failed to start: {proc.stderr.strip()}"
+            )
+
     def ensure_running(
         self, *, startup_timeout: float = DEFAULT_STARTUP_TIMEOUT_SECONDS,
     ) -> str:
-        """Ensure the control plane + shared gateway are up; return the host
-        control-plane URL. Idempotent — a healthy control plane running
-        current code and a running gateway are left untouched. Raises
-        `OrchestratorStartError` on timeout."""
-        gateway = self._gateway()
-        gateway.ensure_built()  # rebuild the bundle image on a source change
-        gateway.ensure_running()  # creates the shared network + (re)starts gateway
+        """Ensure the infra container (control plane + gateway) is up; return
+        the host control-plane URL. Idempotent — a healthy container on current
+        source is left untouched. Raises `OrchestratorStartError` on timeout."""
+        self._build_images()
 
-        # Recreate the orchestrator container only when its bind-mounted
-        # source has actually changed since it started — its Python process
-        # loaded that code at startup and won't reload, so a stale container
-        # would keep running OLD control-plane code. Recreating on *every*
-        # launch (the prior behaviour) would drop every other active
-        # bottle's in-memory egress tokens each time a new bottle starts,
-        # since the orchestrator process holds them only in memory (#381).
         current_hash = source_hash(self._repo_root)
-        if self.is_healthy() and self._orchestrator_source_current(current_hash):
+        if self.is_healthy() and self._infra_source_current(current_hash):
             return self.url
 
-        self._ensure_orchestrator_image()
-        log.info(
-            "starting orchestrator container",
-            context={"name": self._orchestrator_name},
-        )
-        self._run_orchestrator_container(current_hash)
+        log.info("starting infra container", context={"name": self._infra_name})
+        self._run_infra_container(current_hash)
 
         deadline = time.monotonic() + startup_timeout
         while time.monotonic() < deadline:
             if self.is_healthy():
-                log.info("orchestrator healthy", context={"url": self.url})
+                log.info("infra container healthy", context={"url": self.url})
                 return self.url
             time.sleep(_HEALTH_POLL_SECONDS)
         raise OrchestratorStartError(
-            f"orchestrator at {self.url} did not become healthy within {startup_timeout:g}s"
+            f"infra container at {self.url} did not become healthy within {startup_timeout:g}s"
         )
 
     def stop(self) -> None:
-        """Remove the orchestrator + gateway containers (idempotent)."""
-        run_docker(["docker", "rm", "--force", self._orchestrator_name])
-        self._gateway().stop()
+        """Remove the infra container (idempotent)."""
+        run_docker(["docker", "rm", "--force", self._infra_name])
 
 
 __all__ = [
     "OrchestratorService",
     "OrchestratorStartError",
-    "ORCHESTRATOR_NAME",
+    "INFRA_NAME",
+    "INFRA_IMAGE",
+    "INFRA_SOURCE_HASH_LABEL",
     "ORCHESTRATOR_IMAGE",
     "DEFAULT_PORT",
     "DEFAULT_STARTUP_TIMEOUT_SECONDS",
+    "source_hash",
 ]

bot_bottle/orchestrator/registry.py

@@ -113,6 +113,22 @@ _MIGRATIONS = TableMigrations(
         # egress allowlist / routes / git config selected by source IP. The
         # multi-tenant gateway resolves it per request via `attribute`.
         "ALTER TABLE orchestrator_bottles ADD COLUMN policy TEXT NOT NULL DEFAULT ''",
+        # v4 — per-bottle encrypted egress secrets (PRD prd-new-secret-provider).
+        # One row per env-var: key (env-var name) is plaintext for auditing;
+        # value is the encrypted token string. The encryption key (ENV_VAR_SECRET)
+        # lives only in the agent's environment — a row alone cannot recover the
+        # credential.
+        """
+        CREATE TABLE IF NOT EXISTS bottled_agent_secrets (
+            bottled_agent_id TEXT NOT NULL,
+            key              TEXT NOT NULL,
+            value            TEXT NOT NULL,
+            type             TEXT NOT NULL DEFAULT 'injected_env_var'
+        )
+        """,
+        # 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)",
     ],
 )
 
@@ -326,6 +342,57 @@ class RegistryStore(DbStore):
             return None
         return rec
 
+    # --- encrypted egress secret store ------------------------------------
+
+    def store_agent_secrets(
+        self,
+        bottle_id: str,
+        encrypted_values: dict[str, str],
+        secret_type: str = "injected_env_var",
+    ) -> None:
+        """Replace all stored secrets for *bottle_id* with *encrypted_values*
+        (env-var name → encrypted ciphertext). Deletes then re-inserts so a
+        re-registration is always consistent with the current token set."""
+        with self._connection() as conn:
+            conn.execute(
+                "DELETE FROM bottled_agent_secrets "
+                "WHERE bottled_agent_id = ? AND type = ?",
+                (bottle_id, secret_type),
+            )
+            conn.executemany(
+                "INSERT INTO bottled_agent_secrets "
+                "(bottled_agent_id, key, value, type) VALUES (?, ?, ?, ?)",
+                [(bottle_id, k, v, secret_type) for k, v in encrypted_values.items()],
+            )
+        self._chmod()
+
+    def get_agent_secrets(
+        self,
+        bottle_id: str,
+        secret_type: str = "injected_env_var",
+    ) -> dict[str, str]:
+        """Return {env_var_name: encrypted_value} for *bottle_id*, or {} if none."""
+        with self._connection() as conn:
+            rows = conn.execute(
+                "SELECT key, value FROM bottled_agent_secrets "
+                "WHERE bottled_agent_id = ? AND type = ?",
+                (bottle_id, secret_type),
+            ).fetchall()
+        return {row[0]: row[1] for row in rows}
+
+    def delete_agent_secrets(
+        self,
+        bottle_id: str,
+        secret_type: str = "injected_env_var",
+    ) -> None:
+        """Remove all stored secrets for *bottle_id* (e.g. on teardown)."""
+        with self._connection() as conn:
+            conn.execute(
+                "DELETE FROM bottled_agent_secrets "
+                "WHERE bottled_agent_id = ? AND type = ?",
+                (bottle_id, secret_type),
+            )
+
 
 __all__ = [
     "BottleRecord",

bot_bottle/orchestrator/rotate_ca.py

@@ -0,0 +1,62 @@
+"""Rotate the shared gateway's mitmproxy CA (issue #450).
+
+    python -m bot_bottle.orchestrator.rotate_ca
+
+A deliberate CA rollover has two halves: drop the *persisted* CA so a fresh one
+is minted, and drop the *running* gateway so its mitmproxy (which holds the old
+CA in memory) is replaced. This one-shot command does both:
+
+  1. Delete the persisted CA under the host gateway-CA dir — the next gateway
+     start generates a new one (mitmproxy reuses an existing CA, generates only
+     when absent).
+  2. Force-remove the infra / standalone-gateway containers so the stale
+     in-memory CA is gone; the next bottle launch's idempotent `ensure_running`
+     brings the gateway back up and mints the fresh CA.
+
+It does NOT re-provision the new CA into already-running bottles — those must be
+re-attached so they install the new trust anchor. Rotation is thus an explicit,
+operator-driven action with a brief egress interruption, not an automatic one.
+"""
+
+from __future__ import annotations
+
+import sys
+from pathlib import Path
+
+from ..docker_cmd import run_docker
+from ..paths import host_gateway_ca_dir
+from .gateway import GATEWAY_NAME, rotate_gateway_ca
+from .lifecycle import INFRA_NAME
+
+# The containers whose mitmproxy would still be serving the old CA from memory:
+# the consolidated infra container and the standalone per-host gateway.
+_GATEWAY_CONTAINERS = (INFRA_NAME, GATEWAY_NAME)
+
+
+def _out(msg: str) -> None:
+    sys.stdout.write(f"rotate-ca: {msg}\n")
+
+
+def main(argv: list[str] | None = None) -> int:
+    del argv  # no flags — a single deliberate action
+    ca_dir: Path = host_gateway_ca_dir()
+    removed = rotate_gateway_ca(ca_dir)
+    if removed:
+        _out(f"removed {len(removed)} CA file(s) from {ca_dir}")
+    else:
+        _out(f"no persisted CA under {ca_dir}; a fresh one is minted on next start")
+
+    # Drop any running gateway so its in-memory (now-stale) CA is replaced on
+    # the next launch. `rm --force` on an absent name is a tolerated no-op.
+    for name in _GATEWAY_CONTAINERS:
+        proc = run_docker(["docker", "rm", "--force", name])
+        if proc.returncode == 0 and proc.stdout.strip():
+            _out(f"removed running container {name}")
+
+    _out("done — the next bottle launch remints the CA; re-attach bottles to "
+         "install the new trust anchor")
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

bot_bottle/orchestrator/secret_store.py

@@ -0,0 +1,94 @@
+"""Symmetric encryption for per-bottle egress secrets (PRD prd-new-secret-provider).
+
+Each agent receives a random ENV_VAR_SECRET at startup — passed as an env var,
+never logged or persisted. The host uses this key to encrypt each egress auth
+token value before writing it to the bottled_agent_secrets table; the DB rows
+(ciphertext, plaintext env-var name) without the key are insufficient to
+recover the credentials.
+
+On orchestrator restart the in-memory token map is lost. The host-side
+reattachment path reads ENV_VAR_SECRET from the running agent container via
+``docker exec … printenv ENV_VAR_SECRET`` and posts it to
+``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).
+
+    keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big"))
+    ciphertext_i      = plaintext_i XOR keystream_block_i[:len(plaintext_i)]
+"""
+
+from __future__ import annotations
+
+import base64
+import hashlib
+import hmac
+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
+
+# Env-var name the agent container receives at startup.
+ENV_VAR_SECRET_NAME = "ENV_VAR_SECRET"
+
+
+def new_env_var_secret() -> str:
+    """Generate a fresh ENV_VAR_SECRET: 32 random bytes as URL-safe base64."""
+    return base64.urlsafe_b64encode(secrets.token_bytes(_KEY_BYTES)).rstrip(b"=").decode()
+
+
+def _b64dec(s: str) -> bytes:
+    return base64.urlsafe_b64decode(s + "=" * (-len(s) % 4))
+
+
+def _keystream(key: bytes, nonce: bytes, block_index: int) -> bytes:
+    return hmac.new(
+        key, nonce + block_index.to_bytes(4, "big"), hashlib.sha256
+    ).digest()
+
+
+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
+    the ``bottled_agent_secrets.value`` column."""
+    key = _b64dec(secret_b64)
+    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)]
+        ct.extend(p ^ k for p, k in zip(chunk, ks))
+    return base64.urlsafe_b64encode(nonce + bytes(ct)).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)."""
+    key = _b64dec(secret_b64)
+    try:
+        blob = _b64dec(blob_b64)
+    except Exception as exc:
+        raise ValueError(f"invalid ciphertext blob: {exc}") from exc
+    if len(blob) < _NONCE_BYTES:
+        raise ValueError("ciphertext blob too short")
+    nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]
+    pt = bytearray()
+    for i in range(0, len(ciphertext), _BLOCK):
+        chunk = ciphertext[i : i + _BLOCK]
+        ks = _keystream(key, nonce, i)[: len(chunk)]
+        pt.extend(c ^ k for c, k in zip(chunk, ks))
+    try:
+        return bytes(pt).decode()
+    except UnicodeDecodeError as exc:
+        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"]

bot_bottle/orchestrator/service.py

@@ -87,13 +87,22 @@ class Orchestrator:
         metadata: str = "",
         policy: str = "",
         tokens: dict[str, str] | None = None,
+        env_var_secret: str = "",
     ) -> BottleRecord:
         """Register a bottle (with its gateway policy + in-memory egress auth
         tokens) and broker its launch. Rolls the registry entry back if the
-        launch doesn't take, so a failure leaves no orphan."""
+        launch doesn't take, so a failure leaves no orphan.
+
+        When *env_var_secret* is provided alongside *tokens*, the token values
+        are also encrypted and written to ``bottled_agent_secrets`` so they can
+        survive an orchestrator restart (see ``reprovision_from_secret``)."""
         rec = self.registry.register(source_ip, metadata=metadata, policy=policy)
         if tokens:
             self._tokens[rec.bottle_id] = dict(tokens)
+            if env_var_secret:
+                from .secret_store import encrypt_value
+                encrypted = {k: encrypt_value(env_var_secret, v) for k, v in tokens.items()}
+                self.registry.store_agent_secrets(rec.bottle_id, encrypted)
         req = LaunchRequest(
             op="launch",
             bottle_id=rec.bottle_id,
@@ -284,6 +293,26 @@ class Orchestrator:
             ))
         return True, ""
 
+    # --- secret reprovision -----------------------------------------------
+
+    def reprovision_from_secret(self, bottle_id: str, env_var_secret: str) -> bool:
+        """Re-inject a bottle's egress tokens from its ENV_VAR_SECRET.
+
+        Reads the encrypted rows from ``bottled_agent_secrets``, decrypts each
+        value with *env_var_secret*, and restores ``_tokens[bottle_id]``.
+        Returns True on success, False when no stored secrets exist for this
+        bottle or decryption fails (wrong key / corrupt data)."""
+        from .secret_store import decrypt_value
+        encrypted = self.registry.get_agent_secrets(bottle_id)
+        if not encrypted:
+            return False
+        try:
+            self._tokens[bottle_id] = {k: decrypt_value(env_var_secret, v)
+                                       for k, v in encrypted.items()}
+        except ValueError:
+            return False
+        return True
+
     # --- consolidated gateway ----------------------------------------------
 
     def ensure_gateway(self) -> None:

bot_bottle/paths.py

@@ -33,6 +33,13 @@ HOST_DB_FILENAME = "bot-bottle.db"
 CONTROL_PLANE_TOKEN_FILENAME = "control-plane-token"
 CONTROL_PLANE_TOKEN_ENV = "BOT_BOTTLE_CONTROL_PLANE_TOKEN"
 
+# The host directory holding the gateway's persistent mitmproxy CA. Bind-mounted
+# into the infra/gateway container at mitmproxy's confdir so the self-generated
+# CA survives container recreation — every agent installs this one CA to trust
+# the shared gateway's TLS interception, so it must not rotate on restart. See
+# host_gateway_ca_dir() for why this is a host bind-mount, not a named volume.
+GATEWAY_CA_DIRNAME = "gateway-ca"
+
 
 def bot_bottle_root() -> Path:
     """The app data root — `$BOT_BOTTLE_ROOT` if set, else `~/.bot-bottle`."""
@@ -59,6 +66,23 @@ def host_db_dir() -> Path:
     return db_dir
 
 
+def host_gateway_ca_dir() -> Path:
+    """The directory holding the gateway's persistent mitmproxy CA, created if
+    missing. Backends bind-mount this into the infra/gateway container at
+    mitmproxy's confdir so the CA persists across container recreation.
+
+    A host bind-mount under the app-data root — deliberately NOT a Docker
+    named volume. A named volume survives `docker rm` but is silently wiped by
+    `docker volume prune` / `docker system prune --volumes` during routine host
+    maintenance; the gateway then mints a fresh CA that every already-running
+    bottle distrusts, failing the TLS handshake even after it reconnects to the
+    moved gateway (issue #450). A path under the root docker never prunes it,
+    and it stays directly inspectable + rotatable from the host."""
+    ca_dir = bot_bottle_root() / GATEWAY_CA_DIRNAME
+    ca_dir.mkdir(parents=True, exist_ok=True)
+    return ca_dir
+
+
 def host_control_plane_token() -> str:
     """The per-host control-plane secret, minted (256-bit, url-safe) and
     persisted 0600 on first use, then reused.
@@ -94,8 +118,10 @@ __all__ = [
     "HOST_DB_FILENAME",
     "CONTROL_PLANE_TOKEN_FILENAME",
     "CONTROL_PLANE_TOKEN_ENV",
+    "GATEWAY_CA_DIRNAME",
     "bot_bottle_root",
     "host_db_path",
     "host_db_dir",
+    "host_gateway_ca_dir",
     "host_control_plane_token",
 ]

bot_bottle/store_manager.py

@@ -6,9 +6,11 @@ from pathlib import Path
 
 try:
     from .audit_store import AuditStore
+    from .config_store import ConfigStore
     from .queue_store import QueueStore
 except ImportError:
     from audit_store import AuditStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
+    from config_store import ConfigStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
     from queue_store import QueueStore  # type: ignore[import-not-found]  # pylint: disable=import-error,no-name-in-module
 
 _instance: StoreManager | None = None
@@ -47,11 +49,13 @@ class StoreManager:
         return (
             QueueStore("", self.db_path).is_migrated()
             and AuditStore(self.db_path).is_migrated()
+            and ConfigStore(self.db_path).is_migrated()
         )
 
     def migrate(self) -> None:
         QueueStore("", self.db_path).migrate()
         AuditStore(self.db_path).migrate()
+        ConfigStore(self.db_path).migrate()
 
 
 __all__ = ["StoreManager"]

docs/prds/0070-per-host-orchestrator.md

@@ -312,6 +312,44 @@ reaches over the RPC rather than a shared mount into the VM. WAL on the
 shared DB is therefore a deliberate, tested future change — not enabled ad
 hoc. `sqlite3` itself is stdlib, so "the host needs SQLite" is a non-cost.
 
+### Gateway CA: host-resident, like the DB
+
+The shared gateway bumps TLS with a self-generated mitmproxy CA, and **every
+bottle installs that CA** into its trust store to accept the bumped leaves. So
+the CA is durable per-host state with the same rule as the DB: it must outlive
+any single gateway container, or a restart mints a fresh CA that every
+already-running bottle distrusts — the TLS handshake then fails even after the
+bottle re-resolves and reconnects to the moved gateway (issue #450, a
+re-attachment blocker distinct from #443/#445).
+
+The CA lives on the **host filesystem** at `bot_bottle_root()/gateway-ca`
+(`host_gateway_ca_dir()`), bind-mounted into the container at mitmproxy's
+confdir. This is deliberately a host bind-mount, **not a container-runtime
+named volume**: a named volume survives ordinary container removal but can be
+silently wiped by Docker's or Apple Container's volume-prune commands during
+routine host maintenance, which is exactly how the ephemeral-CA symptom shows
+up in practice. A path under the app-data root is not managed or pruned by the
+container runtime, and stays directly inspectable and rotatable from the host.
+mitmproxy reuses an existing CA and generates one only on first run, so the
+bind-mount alone gives
+"adopt-existing, generate-on-first-run" for free.
+
+The macOS backend uses the same host-resident CA directory and bind-mounts it
+into the consolidated Apple infra container. Its `bot-bottle-mac-db` named
+volume remains container-only because that prevents incoherent cross-kernel
+SQLite locking, but the CA is deliberately not stored there: Apple Container
+also has a `container volume prune` operation, and the named volume is
+temporarily unreferenced while the infra container is recreated. Keeping the
+CA on the host makes both ordinary recreation and volume pruning safe.
+
+**Deliberate rollover** is the explicit inverse: `rotate_gateway_ca()` removes
+the persisted CA material so the next start remints it, and the
+`python -m bot_bottle.orchestrator.rotate_ca` one-shot wires that together with
+dropping the running gateway container (whose mitmproxy still holds the old CA
+in memory). Rotation does not auto-re-provision the new CA into running bottles
+— those re-attach to install the new anchor — so it is an operator action with
+a brief egress interruption, never an implicit one.
+
 ## Sequencing
 
 Jump straight to the **virtualized** end state (not a host-daemon stepping

docs/prds/prd-new-ci-artifact-coverage.md

@@ -0,0 +1,110 @@
+# PRD prd-new: CI artifact-based coverage and local Firecracker candidate flow
+
+- **Status:** Active
+- **Author:** Claude
+- **Created:** 2026-07-21
+- **Issue:** #446
+
+## Summary
+
+Restructure the CI test pipeline to run each test suite exactly once, upload
+small `.coverage.*` artifacts, and combine them in a lightweight aggregation
+job. Move the infra build onto the KVM runner so the ~194 MB rootfs never
+crosses the network for PRs. On main-branch pushes, publish the byte-identical
+rootfs that was tested.
+
+## Motivation
+
+The prior pipeline had two redundant costs:
+
+1. **Duplicate artifact transfers.** `build-infra` (ubuntu-latest) built and
+   uploaded the ~194 MB rootfs; `integration-firecracker` downloaded it; the
+   `coverage` job downloaded it a second time. Combined download overhead: ~83
+   seconds per run, plus the ~70-second upload.
+
+2. **Duplicate test execution.** `integration-firecracker` ran the Firecracker
+   integration suite; `coverage` ran the entire unit + integration suite again
+   on the same KVM runner to collect coverage data. Every line of Firecracker
+   code was tested twice per CI run.
+
+## Goals
+
+- Each test suite (unit, integration-docker, integration-firecracker) executes
+  exactly once per workflow run.
+- PRs incur no large artifact transfers — the rootfs stays on the KVM runner.
+- Main-branch pushes publish a byte-for-byte identical rootfs to the one that
+  passed the integration tests.
+- Concurrent workflow runs cannot cross-publish candidates (naturally enforced
+  by Gitea Actions' per-run artifact scoping).
+- Failed or cancelled runs block publication (enforced by the `needs:` chain on
+  `publish-infra`).
+
+## Non-goals
+
+- Changing test semantics or the coverage policy (ADR 0004).
+- Removing the KVM runner guard on `integration-firecracker` and `coverage`.
+- Changing how `publish_infra.py` builds or uploads the rootfs.
+
+## Design
+
+### Job graph
+
+```
+unit ──────────────────────────────────┐
+integration-docker ────────────────────┤──► coverage ──► publish-infra (main only)
+integration-firecracker (KVM) ─────────┘
+```
+
+### `unit`
+
+Unchanged except: `coverage run` writes `--data-file=.coverage.unit`; the file
+is uploaded as the `coverage-unit` artifact.
+
+### `integration-docker`
+
+Adds a `coverage` install step. `coverage run` writes `--data-file=.coverage.docker`;
+the file is uploaded as `coverage-docker`.
+
+### `integration-firecracker` (KVM runner)
+
+Replaces the old `stage-firecracker-inputs` → `build-infra` → download chain:
+
+1. Builds the infra candidate locally with
+   `BOT_BOTTLE_FC_DROPBEAR=/var/cache/bot-bottle-fc/dropbear`.
+2. Boots the candidate and runs integration tests with coverage, writing
+   `.coverage.firecracker`.
+3. Uploads the small `coverage-firecracker` artifact unconditionally.
+4. On main-branch pushes only, uploads the rootfs as `infra-candidate` and the
+   dropbear as `firecracker-inputs` so `publish-infra` can verify and publish
+   the byte-identical artifact.
+
+### `coverage`
+
+Moves from a KVM runner to `ubuntu-latest`. No tests are re-executed:
+
+1. Downloads `coverage-unit`, `coverage-docker`, and `coverage-firecracker`.
+2. Runs `scripts/coverage.sh aggregate critical`, which calls
+   `coverage combine` then `coverage report`.
+3. Runs the diff-coverage gate (`scripts/diff_coverage.py`).
+
+Coverage files use `relative_files = True` (`.coveragerc`) so they combine
+cleanly across runners with different absolute workspace paths.
+
+### `publish-infra`
+
+Depends on all four predecessor jobs (unchanged gate). Downloads `infra-candidate`
+and `firecracker-inputs` that were uploaded by `integration-firecracker` on
+main — the same byte sequence that passed the integration tests.
+
+### Eliminated jobs
+
+- `stage-firecracker-inputs`: existed only to copy the dropbear to ubuntu-latest
+  for `build-infra`. No longer needed.
+- `build-infra`: the infra candidate is now built on the KVM runner in
+  `integration-firecracker`.
+
+### Script changes
+
+`scripts/coverage.sh` gains an `aggregate` mode (`coverage.sh aggregate [critical]`)
+that combines pre-existing `.coverage.*` files instead of re-running tests.
+The existing run mode (`coverage.sh [critical]`) is preserved for local dev.

docs/prds/prd-new-claude-forward-host-credentials.md

@@ -0,0 +1,146 @@
+# PRD prd-new: Claude forward_host_credentials
+
+- **Status:** Draft
+- **Author:** claude
+- **Created:** 2026-07-01
+- **Issue:** #325
+
+## Summary
+
+Add `agent_provider.forward_host_credentials: true` support for the
+`claude` template, mirroring the existing Codex flow. When enabled,
+bot-bottle reads the host's Claude OAuth session key from
+`~/.claude/.credentials.json` at launch, forwards it only to the egress sidecar,
+and injects a placeholder `CLAUDE_CODE_OAUTH_TOKEN` into the agent so
+Claude Code starts without ever seeing the real credential.
+
+## Problem
+
+Running a Claude agent in a container today requires the operator to
+manually extract a long-lived OAuth token (`claude setup-token`), export
+it as `BOT_BOTTLE_CLAUDE_OAUTH_TOKEN`, and reference it explicitly in
+the manifest with `agent_provider.auth_token:
+"BOT_BOTTLE_CLAUDE_OAUTH_TOKEN"`. This is a two-step manual ceremony
+that is easy to skip or do incorrectly.
+
+The host already stores a valid Claude session in `~/.claude/.credentials.json`
+after `claude login`. Codex already automates an
+equivalent extraction from `~/.codex/auth.json`. There is no reason
+Claude bottles cannot do the same.
+
+## Goals / Success Criteria
+
+- A Claude bottle with `forward_host_credentials: true` in the manifest
+  uses the host's `~/.claude/.credentials.json` session key at launch with no
+  additional operator steps.
+- The agent container receives only `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder`
+  — never the real token.
+- The real session key lives only in the egress sidecar's environment.
+- Missing, malformed, or expired host Claude auth fails launch with a
+  clear operator-facing message.
+- Existing `auth_token` behavior is unchanged.
+- `forward_host_credentials: true` is rejected in the manifest when both
+  `auth_token` and `forward_host_credentials` are set, since they serve
+  the same purpose.
+
+## Non-goals
+
+- Refreshing Claude OAuth tokens in the sidecar.
+- Writing a dummy `~/.claude.json` auth state to the agent (unlike the
+  Codex flow, Claude Code reads its credential from `CLAUDE_CODE_OAUTH_TOKEN`
+  in env, not from an auth file — no guest-side auth marker is needed).
+- Supporting `forward_host_credentials` for providers other than `codex`
+  and `claude`.
+
+## Design
+
+### Manifest schema
+
+```yaml
+agent_provider:
+  template: claude
+  forward_host_credentials: true
+```
+
+Rejects in manifest validation when:
+- Template is not `codex` or `claude`.
+- Both `auth_token` and `forward_host_credentials` are set.
+
+### Host auth extraction (`contrib/claude/claude_auth.py`)
+
+Claude Code credential storage varies by platform:
+
+- **Linux**: `~/.claude/.credentials.json`
+- **macOS**: macOS Keychain, service `"Claude Code-credentials"`
+  (the file path is tried first; Keychain is the fallback when the file
+  is absent)
+
+`~/.claude.json` contains only UI state and profile metadata — no token.
+
+The credentials JSON schema (same whether from file or Keychain):
+
+```json
+{
+  "claudeAiOauth": {
+    "accessToken": "<access-token>",
+    "refreshToken": "<refresh-token>",
+    "expiresAt": 1748276587173,
+    "scopes": ["user:inference", "user:profile"]
+  }
+}
+```
+
+`expiresAt` is in **milliseconds** (not seconds).
+
+At prepare/launch time, when `forward_host_credentials: true`:
+
+1. Try `~/.claude/.credentials.json`; on macOS, if absent, run
+   `security find-generic-password -s "Claude Code-credentials" -w`
+   and parse its stdout as JSON.
+2. Require a `claudeAiOauth` dict.
+3. Require a non-empty `claudeAiOauth.accessToken` string.
+4. If `claudeAiOauth.expiresAt` is present, divide by 1000 and require
+   the result to be in the future.
+5. Return only the access token to the launch path.
+
+Errors name the missing or invalid condition and point the operator at
+`claude login`, without printing token values.
+
+### Egress route
+
+When `forward_host_credentials: true`:
+
+- Provision the session key in `provisioned_env` under
+  `BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN` (new constant in `egress.py`).
+- Set up the `api.anthropic.com` egress route with `auth_scheme: Bearer`
+  and `token_ref: BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN`.
+- Set `CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder` in the agent env and
+  add it to `hidden_env_names`.
+
+No dummy auth file and no `verify` step are needed — Claude Code reads
+the credential from the env var, not from a file.
+
+### Constants
+
+- `CLAUDE_HOST_CREDENTIAL_TOKEN_REF = "BOT_BOTTLE_CLAUDE_HOST_ACCESS_TOKEN"`
+  in `egress.py` (alongside the existing `CODEX_HOST_CREDENTIAL_TOKEN_REF`).
+- `CLAUDE_HOST_CREDENTIAL_HOSTS = ("api.anthropic.com",)` in
+  `agent_provider.py` (alongside the existing `CODEX_HOST_CREDENTIAL_HOSTS`).
+
+### Data flow
+
+```
+Host ~/.claude/.credentials.json  →  bot-bottle launch
+                            │
+                            ├──► egress sidecar env (real token only)
+                            │
+                            └──► agent env: CLAUDE_CODE_OAUTH_TOKEN=egress-placeholder
+
+Agent  →  HTTPS to api.anthropic.com (via egress)
+Egress →  injects Authorization: Bearer <real token>
+Egress →  forwards to api.anthropic.com
+```
+
+## Open questions
+
+None — the Codex precedent makes the design clear.

docs/prds/prd-new-consolidate-docker-infra-backend.md

@@ -0,0 +1,156 @@
+# PRD prd-new: Consolidate infra backend for Docker
+
+- **Status:** Active
+- **Author:** Claude
+- **Created:** 2026-07-20
+- **Issue:** #431
+
+## Summary
+
+The Docker backend runs two containers — `bot-bottle-orch-gateway` (gateway
+data plane) and `bot-bottle-orchestrator` (control plane) — where the
+macOS and Firecracker backends already run a single combined infra
+unit. This PRD collapses Docker to the same model: one `bot-bottle-infra`
+container running both processes under the `gateway_init` supervise tree, a
+restructured `Dockerfile.infra` as the shared gateway+orchestrator base,
+and a handful of extracted shared utilities (CA cert polling, teardown
+sequence, launch skeleton) that are currently duplicated across all three
+`consolidated_launch.py` files.
+
+## Goals / success criteria
+
+- Docker backend starts exactly one infra container instead of two.
+- `Dockerfile.infra` is the shared base image (gateway + orchestrator, no
+  buildah); the Firecracker image layers buildah on top of it.
+- The orchestrator process runs under the `gateway_init` supervise tree
+  inside the combined container (one PID-1, one restart/health surface).
+- CA cert polling, the teardown sequence, and the shared launch skeleton
+  (ensure-infra → register → provision → return context) live in a single
+  shared module; all three backends import from it.
+- No functional change to macOS or Firecracker launch paths.
+
+## Non-goals
+
+- Changing per-bottle isolation — agents stay one-VM/container-each.
+- Consolidating transport implementations (`DockerGatewayTransport`,
+  `AppleGatewayTransport`, `SshGatewayTransport`) — these are already the
+  right abstraction boundary.
+- macOS DHCP-inversion of registration order — irreducible backend
+  difference, stays as-is.
+- Any changes to the orchestrator RPC protocol or the attribution model.
+
+## Design
+
+### Dockerfile restructuring
+
+**Current shape:**
+
+- `Dockerfile.gateway` — data plane (mitmproxy, gitleaks, git, openssh,
+  supervise daemons)
+- `Dockerfile.orchestrator` — control plane (python:3.12-slim + bot_bottle
+  package; stdlib-only, no third-party deps)
+- `Dockerfile.infra` — Firecracker only: `FROM bot-bottle-gateway` +
+  buildah + `COPY --from bot-bottle-orchestrator`
+
+**New shape:**
+
+- `Dockerfile.gateway` — unchanged
+- `Dockerfile.orchestrator` — unchanged (single definition of orchestrator
+  content; both Docker infra and Firecracker infra `COPY --from` it)
+- `Dockerfile.infra` — **shared base**: `FROM bot-bottle-gateway` + `COPY
+  --from bot-bottle-orchestrator` (no buildah — Docker infra image)
+- `Dockerfile.infra.fc` — Firecracker only: `FROM bot-bottle-infra` +
+  buildah/crun/netavark/aardvark-dns (layered on the shared base, same net
+  result as today)
+
+The comment in `Dockerfile.infra` that says "the docker backend keeps
+orchestrator + gateway as separate images; this combined image exists only
+for the Firecracker single-VM cut" is removed.
+
+### Orchestrator in the supervise tree
+
+`gateway_init` already supervises the data-plane daemons (egress, git-http,
+supervise-MCP). The orchestrator control plane is added as another supervised
+process: `python3 -m bot_bottle.orchestrator --host 0.0.0.0 --port <port>
+--broker stub`.
+
+The orchestrator source is bind-mounted (`/app` → repo root, as today) so
+dev live-reload still works. `source_hash`-based container recreation in
+`OrchestratorService.ensure_running` continues to apply — a code change
+recreates the combined infra container, which bounces both gateway and
+orchestrator. This is acceptable: the docker backend is a dev/legacy target
+where in-flight egress connections across a code deploy are not a hard
+requirement.
+
+### `OrchestratorService` changes
+
+`OrchestratorService` currently starts two containers in sequence: gateway
+first (`DockerGateway.ensure_running`), then orchestrator. After this PRD:
+
+- Single `docker run` of `bot-bottle-infra:latest`
+- Container name: `bot-bottle-infra` (replaces `bot-bottle-orch-gateway` +
+  `bot-bottle-orchestrator`)
+- Published ports: `127.0.0.1:{host_port}:8099` for the control plane
+  (`gateway_init` listens on a fixed internal port 8099; the caller-chosen
+  host port maps to it)
+- Bind mounts: repo root + host root (same as today)
+- `DockerGateway` becomes an implementation detail of `OrchestratorService`
+  rather than a separately started container; the gateway image name
+  (`GATEWAY_IMAGE`) is no longer referenced at runtime, only at build time
+  for the `Dockerfile.infra` base
+
+The `_gateway()` / `ensure_running` two-step in `OrchestratorService` is
+replaced by a single `_run_infra_container()`.
+
+### Shared backend utilities
+
+Three items are duplicated across
+`backend/docker/consolidated_launch.py`,
+`backend/macos_container/consolidated_launch.py`, and
+`backend/firecracker/consolidated_launch.py`:
+
+1. **CA cert polling loop** — `deadline = time.monotonic() + timeout; while
+   ...: try fetch CA; sleep` — extracted to
+   `backend/consolidated_util.py:poll_ca_cert(transport, *, timeout)`.
+
+2. **Teardown sequence** — `OrchestratorClient(url).teardown_bottle(id)` +
+   `deprovision_git_gate(transport, id)` — extracted to
+   `backend/consolidated_util.py:teardown_consolidated(url, transport,
+   bottle_id)`.
+
+3. **Launch skeleton** — all three follow: ensure-infra → allocate/register
+   → provision git-gate → fetch CA cert → return launch context. The macOS
+   inversion (agent starts before registration, source IP from DHCP) is the
+   only deviation. Extract a shared `_provision_bottle(transport, bottle_id,
+   plan, orchestrator_url)` helper covering the register → provision →
+   return-token steps; the backends keep their own `launch_consolidated`
+   wrappers for the before/after (infra-ensure + agent-start + IP
+   allocation), calling the shared helper.
+
+The new `backend/consolidated_util.py` module holds only backend-neutral,
+transport-agnostic logic. All three backends import from it.
+
+## Implementation chunks
+
+1. **(this PR)** Dockerfile restructuring: rename current `Dockerfile.infra`
+   content to `Dockerfile.infra.fc`; write new `Dockerfile.infra` as
+   gateway+orchestrator base. Update Firecracker image-build references from
+   `Dockerfile.infra` → `Dockerfile.infra.fc`.
+
+2. Add orchestrator process to `gateway_init` supervise tree.
+
+3. Collapse `OrchestratorService` to a single-container start; rename
+   container from `bot-bottle-orch-gateway`/`bot-bottle-orchestrator` →
+   `bot-bottle-infra`; update image name constant.
+
+4. Extract `backend/consolidated_util.py` with `poll_ca_cert`,
+   `teardown_consolidated`, and `_provision_bottle`; update all three
+   `consolidated_launch.py` files to import from it.
+
+5. Update tests that reference the old container names or two-container
+   startup sequence.
+
+## Open questions
+
+None — the supervise-tree approach and shared Dockerfile layering were
+confirmed in issue #431.

docs/prds/prd-new-modernize-built-in-agent-images.md

@@ -0,0 +1,47 @@
+# PRD prd-new: Modernize built-in agent images
+
+- **Status:** Draft
+- **Author:** Codex
+- **Created:** 2026-07-21
+- **Issue:** #451
+
+## Summary
+
+Keep every built-in agent provider on Debian's current stable release and make
+Podman available inside each image. This gives agents a consistent, modern
+userspace and an OCI container tool without requiring per-project setup.
+
+## Problem
+
+The Claude, Codex, and Pi images inherit the generic `node:22-slim` tag. That
+tag does not state which Debian release the project supports and currently
+leaves the images on the older Bookworm release. None of the built-in images
+installs Podman, so tasks that need to inspect or build OCI images must first
+modify the bottle or cannot run at all.
+
+## Goals / success criteria
+
+- Every Dockerfile under `bot_bottle/contrib/*/Dockerfile` explicitly inherits
+  `node:22-trixie-slim`, based on Debian 13 (the current stable release).
+- Every built-in agent image installs Podman from Debian stable.
+- Every built-in agent image retains an SSH client for Git-over-SSH workflows.
+- The non-root agent user owns a traversable XDG Git configuration directory,
+  so Git can load bot-bottle's global git-gate rewrites without permission
+  errors.
+- A shared test enforces both requirements for current and future built-in
+  providers.
+
+## Non-goals
+
+- Configuring privileged or nested-container execution for bottles.
+- Pinning Podman outside Debian's stable package repository.
+- Changing the Node.js or agent CLI release policy.
+
+## Design
+
+Use the explicit `node:22-trixie-slim` base rather than the floating `slim`
+variant. Install the `podman` package with each image's existing `apt-get`
+dependency layer, so package metadata and caches are still removed in the same
+layer. Treat Debian stable as the Podman stability and update channel; this
+keeps the images stdlib/distribution-first and avoids adding a third-party
+package repository.

docs/research/egress-proxy-oom-on-large-downloads.md

@@ -1,86 +0,0 @@
-# Egress proxy OOMs on large downloads
-
-Found on 2026-07-21 while running the rootless-podman spike
-(`docs/research/rootless-docker-in-apple-container-spike.md`). Recorded
-rather than fixed — the fix is a security-relevant decision, not a
-mechanical patch.
-
-## Summary
-
-A single large HTTPS download through the gateway kills the egress
-proxy. `mitmdump` buffers whole response bodies so the DLP detectors can
-scan them, grows past the gateway container's memory limit, and is
-OOM-killed by the cgroup. Nothing restarts it.
-
-Two properties make this worse than a failed download:
-
-- **The gateway is a per-host singleton.** Every bottle shares it, so
-  one bottle's download takes egress away from all of them.
-- **There is no restart on death.** The gateway supervisor is
-  `while : ; do wait ; done`; a killed daemon stays dead until the infra
-  container is recreated.
-
-So ordinary agent activity — pulling a container image, downloading a
-model or dataset, fetching a large tarball — is a denial of service
-against every other bottle on the host. No malice required, though it is
-trivially reachable on purpose.
-
-## Evidence
-
-Triggered by `docker compose up` pulling `quay.io/fedora/python-312`
-(two layers, ~82MB and ~83MB) inside a bottle. The pull itself
-succeeded; the *next* request failed:
-
-```
-initializing source docker://quay.io/fedora/python-312:latest:
-  pinging container registry quay.io: Get "https://quay.io/v2/":
-  proxyconnect tcp: dial tcp 192.168.128.39:9099: connect: connection refused
-```
-
-From the gateway's `dmesg`:
-
-```
-python3 invoked oom-killer: gfp_mask=0x100cca(GFP_HIGHUSER_MOVABLE), order=0
-oom-kill:constraint=CONSTRAINT_MEMCG,
-  oom_memcg=/container/bot-bottle-mac-infra,
-  task_memcg=/container/bot-bottle-mac-infra,task=mitmdump,pid=118
-Memory cgroup out of memory: Killed process 118 (mitmdump)
-  total-vm:1391936kB, anon-rss:997768kB
-```
-
-~1GB RSS against a 1024MB container. Note the amplification: ~165MB of
-layers produced ~1GB of resident memory, so the buffering is several
-copies deep (encoded body, decoded body, and the text conversion the
-regex detectors scan).
-
-Afterwards the gateway container was still running and healthy-looking —
-orchestrator, supervise, and git-http all alive — with no `mitmdump`
-process at all, and it stayed that way until the container was
-recreated. A liveness check on the container would not have caught this.
-
-## Reproduction
-
-1. Launch any bottle with an egress route to a host serving a large file.
-2. Download >~150MB over HTTPS through the proxy.
-3. `dmesg | grep -i oom` inside `bot-bottle-mac-infra`, and note that no
-   `mitmdump` process remains.
-
-Beware a false negative when checking: truncating the process listing
-(`cut -c1-45`) cuts before the binary name, because `mitmdump` runs as
-`/usr/local/bin/python3.12 /usr/local/bin/mitmdump …`.
-
-## Fix options, not yet chosen
-
-1. **Restart dead daemons.** Smallest change and strictly an
-   improvement: an OOM then degrades one download instead of removing
-   egress for every bottle. Does not stop the OOM.
-2. **Cap the scanned body size.** Above a threshold, stop buffering —
-   either skip the scan or stream it. This is the root-cause fix and a
-   security decision: a size threshold is exactly the hole an exfiltrator
-   would aim for, so "skip above N" trades a DoS for a covert channel.
-   Streaming with a bounded window keeps coverage, at more complexity.
-3. **Raise the gateway's memory limit.** Moves the threshold; does not
-   remove it.
-
-Worth noting that (1) and (2) are complementary — the restart gap is
-worth closing regardless of how the memory behaviour is resolved.

docs/research/rootless-docker-in-apple-container-spike.md

@@ -1,353 +0,0 @@
-# Rootless Docker inside Apple Container bottles
-
-Spike branch: `spike/rootless-docker-macos` (`a4d8461`)
-
-## Summary
-
-**Negative result.** Rootless Docker cannot run inside an Apple
-Container bottle without granting the bottle `CAP_SYS_ADMIN`. This is a
-kernel constraint on writing multi-range `uid_map`, not a packaging gap
-we can close with a better init script, a different base image, or more
-careful `/etc/subuid` handling.
-
-The spike was built on the premise — stated in
-`bot_bottle/backend/macos_container/rootless_docker.py` — that it would
-*"deliberately refuse to compensate for missing prerequisites with outer
-capabilities, a privileged container, or a host Docker socket."* That
-premise is exactly what the experiment falsified. The two ways forward
-are to abandon the premise (add `CAP_SYS_ADMIN` to the bottle, and with
-it most of the isolation the bottle exists to provide) or to abandon
-rootless Docker.
-
-Recommendation: abandon rootless Docker. Podman does not have this
-problem — see [Podman is not blocked by
-this](#podman-is-not-blocked-by-this) below.
-
-## Local environment
-
-Tested on 2026-07-21:
-
-```console
-$ sw_vers
-ProductName:		macOS
-ProductVersion:		26.5.1
-BuildVersion:		25F80
-
-$ container --version
-container CLI version 1.0.0 (build: release, commit: ee848e3)
-
-$ uname -a   # inside the bottle
-Linux ... 6.18.15 #1 SMP Tue Mar 17 01:36:53 UTC 2026 aarch64 GNU/Linux
-```
-
-## The failure
-
-`tests/integration/test_macos_rootless_docker_spike.py` builds the
-image, launches the bottle, and dies in `rootless_docker.start`:
-
-```
-+ exec rootlesskit --net=slirp4netns --mtu=65520 ... dockerd-rootless.sh
-[rootlesskit:parent] error: failed to setup UID/GID map:
-  newuidmap 1100 [0 1000 1 1 100000 65536] failed:
-  newuidmap: write to uid_map failed: Operation not permitted
-```
-
-## Why it fails
-
-Every prerequisite you would normally suspect is present and correct in
-the guest:
-
-| Check | Result |
-| --- | --- |
-| `/usr/bin/newuidmap` | `-rwsr-xr-x root root` — setuid bit intact, survived the OCI export |
-| `/` mount options | `rw,relatime` — **not** `nosuid` |
-| `NoNewPrivs` | `0` |
-| `Seccomp` | `0`, no filters |
-| `/etc/subuid`, `/etc/subgid` | `node:100000:65536` in both |
-| user namespace | `user:[4026531837]`, identical to pid 1 — the *initial* userns |
-| `unshare -U -r true` | succeeds |
-| `/proc/sys/user/max_user_namespaces` | `4505` |
-
-The one thing that is missing is in the capability bounding set that
-Apple Container gives the container:
-
-```
-CapBnd: 00000000a80425fb
-= chown, dac_override, fowner, fsetid, kill, setgid, setuid, setpcap,
-  net_bind_service, net_raw, sys_chroot, mknod, audit_write, setfcap
-```
-
-No `CAP_SYS_ADMIN`. That is the whole story, and the chain is:
-
-1. The kernel's `map_write()` gates writing a `uid_map` on
-   `file_ns_capable(file, ns, CAP_SYS_ADMIN)` — capability over the
-   **new** user namespace, evaluated against the credentials that opened
-   `/proc/<pid>/uid_map`.
-2. `newuidmap` is setuid-root, so it runs with euid 0 — but its
-   capability sets are clamped by the bounding set, which has no
-   `CAP_SYS_ADMIN`.
-3. `cap_capable()` has a shortcut that grants *all* capabilities when
-   the caller's userns is the new namespace's parent **and**
-   `ns->owner == cred->euid`. It does not apply: the namespace was
-   created by `node` (uid 1000) while `newuidmap` runs as euid 0.
-4. So the check falls through to the effective-set test in the initial
-   userns, which fails. `EPERM`.
-
-Note that the single-line unprivileged path (`unshare -U -r`) works
-precisely because it does not go through `newuidmap` and does not need
-`CAP_SYS_ADMIN`. Only the multi-range subuid mapping that rootless
-Docker requires does.
-
-This is the same constraint that makes upstream's `dind-rootless` image
-require `--privileged`. It is not specific to Apple Container, except
-that Apple Container gives us no bounding set that includes
-`CAP_SYS_ADMIN` by default.
-
-## It does work with the capability — which is the point
-
-Adding the capability clears the failure immediately, and exposes one
-further, much smaller blocker: `/dev/net/tun` exists (the kernel has
-tun; `/proc/misc` lists `200 tun`) but Apple Container creates it
-`crw------- root root`, so uid 1000 cannot open it and `slirp4netns`
-fails with `open: Permission denied`. A `chmod 0666 /dev/net/tun` as
-root inside the bottle fixes that, and needs no capability beyond what
-the bottle already has.
-
-With both applied by hand, the daemon comes up completely:
-
-```console
-$ container run --rm -u root --cap-add CAP_SYS_ADMIN \
-    bot-bottle-claude:latest-rootless-docker sh -c '...'
-Server Version: 20.10.24+dfsg1
-Storage Driver: fuse-overlayfs
-Cgroup Driver: none
-Cgroup Version: 2
-API listen on /tmp/rt/docker.sock
-```
-
-So `rootless-docker-init.sh` and `rootless_docker.py` are *correct*.
-The spike did not fail on a bug. It failed on its own premise.
-
-Two secondary findings from that run, relevant if anyone revisits this:
-
-- Debian's `docker.io` package pins Docker **20.10** (EOL), not the 28.x
-  implied by the `docker:28-cli` compose plugin the image copies in.
-- `Cgroup Driver: none` — no resource limits on nested containers.
-
-## Why we should not just add the capability
-
-`CAP_SYS_ADMIN` is close to a superset of "root" in practical terms —
-mount, `pivot_root`, namespace manipulation, and a long tail of
-subsystem-specific powers. Granting it to the agent bottle would
-undercut the containment argument the rest of the backend is built
-around, including the deliberately narrow choices immediately adjacent
-to it in `launch.py` (`--cap-drop CAP_NET_RAW`, no `NET_ADMIN`, a
-host-only agent network). Trading all of that for nested `docker
-compose` is a bad exchange.
-
-## Podman is not blocked by this
-
-Sanity-checked on the same host, same kernel, same runtime, so the
-comparison is apples to apples:
-
-| Scenario | Result |
-| --- | --- |
-| Podman rootless, `/etc/subuid` populated | **Fails identically** — `newuidmap: write to uid_map failed: Operation not permitted` |
-| Podman rootless, no subuid ranges, `--network=host` | **Works**, no added capabilities |
-| Podman rootless, no subuid ranges, default netns, `/dev/net/tun` at `0600` | Fails — `slirp4netns: open("/dev/net/tun"): Permission denied` |
-| Podman rootless, no subuid ranges, default netns, `/dev/net/tun` at `0666` | **Works**, no added capabilities |
-
-The difference is that podman degrades gracefully when no subuid range
-is available: it falls back to a single-UID self-mapping, which an
-unprivileged process may write itself, so `newuidmap` is never invoked
-and `CAP_SYS_ADMIN` is never needed. Docker's rootless mode has no
-equivalent fallback.
-
-The cost of that fallback is real and should be weighed before building
-on it: with a single-UID mapping, every UID inside a nested container
-collapses onto the bottle's own uid 1000. There is no UID separation
-between the agent and anything it runs — `root` in a nested container is
-the agent user outside it. It also requires `ignore_chown_errors` on the
-storage driver. Whether that is acceptable depends on whether the bottle
-boundary (which is unchanged) or the nested-container boundary (which is
-effectively nil) is the one we are relying on.
-
-## What the podman spike then needed
-
-The podman implementation that replaced the Docker one on this branch
-turned up two more device-node blockers of the same shape as
-`/dev/net/tun` — Apple Container creates the node, but 0600 root:root:
-
-- **`/dev/fuse`** — blocks the `fuse-overlayfs` storage driver
-  (`fuse: failed to open /dev/fuse: Permission denied`). Without it the
-  only working driver is `vfs`, which copies whole layers per container.
-- **`/dev/net/tun`** — blocks `slirp4netns`, which rootless podman uses
-  for the default bridge network.
-
-Both are fixed by `chmod 0666` as root inside the bottle, which needs no
-capability the bottle does not already hold. This is categorically
-different from the `CAP_SYS_ADMIN` requirement: it is a permission on a
-node that already exists, not an outer privilege grant.
-
-One design note worth recording: the agent-facing surface stays `docker`
-and `docker compose`, pointed at podman's Docker-compatible API socket
-via `DOCKER_HOST`. Setting `netns="host"` in `containers.conf` does *not*
-propagate through that compat API — stock `docker run` and compose files
-request bridge networking explicitly — so slirp4netns (and therefore the
-`/dev/net/tun` chmod) is required for ordinary compose files to work at
-all. Host networking remains available per-workload via
-`--network=host`.
-
-Verified working in a bottle with zero added capabilities: fuse-overlayfs
-storage, the compat API socket, `docker run` on both bridge and host
-networking, and published ports.
-
-### Nested pulls collide with our own egress DLP
-
-The first live run got podman up and `docker compose` running, then
-failed on the image pull:
-
-```
-web Pulling
-initializing source docker://python:3.12-alpine: reading manifest ...
-StatusCode: 403, egress DLP: Generic Bearer JWT found in body
-```
-
-This is bot-bottle's own egress scanner, not a podman problem. The
-Docker registry auth flow carries a bearer JWT *by protocol*, and the
-`token_patterns` detector's `Generic Bearer JWT` rule
-(`Bearer\s+[A-Za-z0-9._\-]{50,}`) matches it on every pull. Any bottle
-that pulls images will hit this.
-
-The fix is per-route detector scoping, which the egress config already
-supports — drop `token_patterns` on the registry hosts and keep
-`known_secrets`:
-
-```json
-{"host": "registry-1.docker.io",
- "dlp": {"outbound_detectors": ["known_secrets"]}}
-```
-
-That is the right trade rather than a grudging one: `known_secrets`
-matches the bottle's *actual* credential values, so real exfil through a
-registry host is still caught. `token_patterns` on a registry route only
-ever produces protocol noise.
-
-Worth generalising later: any manifest enabling `docker_access` needs
-this on its registry routes, so it probably belongs in a shared
-registry-route snippet rather than being copy-pasted per bottle.
-
-### And then registry auth collides with the Authorization strip
-
-With DLP scoped, the pull failed differently: `unauthorized:
-authentication required`. This one is architectural.
-
-`egress_addon.py` strips agent-set `Authorization` unconditionally
-before forwarding — deliberately, so an agent cannot smuggle a
-credential out in a header the DLP detectors don't recognise. A route
-may carry gateway-injected auth instead, but only from a *static* token
-in an env var (`auth_scheme` + `token_env`).
-
-Docker registry auth doesn't fit that shape. The client fetches a
-short-lived, per-repository-scope bearer token from `auth.docker.io` and
-presents it to `registry-1.docker.io`. There is no static token to
-inject, and the token the client legitimately obtained is stripped.
-
-Measured inside a bottle, by hand:
-
-| Step | Result |
-| --- | --- |
-| Fetch token from `auth.docker.io` | 200, 5409-byte token body |
-| Manifest request **with** that valid token | 401 |
-| Manifest request with **no** Authorization | 401 — identical |
-
-A valid token behaves exactly like sending none, which is direct
-evidence the header never arrives. Any nested-container workflow that
-pulls from a registry is blocked on this, so it is not a detail that can
-be deferred: pulling base images is most of what nested containers are
-for.
-
-### Registries that skip the token dance work today
-
-Not every registry needs the stripped header. Measured directly:
-
-| Registry | Manifest request with no `Authorization` |
-| --- | --- |
-| `quay.io` | 200 |
-| `mcr.microsoft.com` | 200 |
-| `registry.k8s.io` | 307 (redirect, no auth) |
-| `ghcr.io` | 401 |
-| `registry-1.docker.io` | 401 |
-
-So "just add the registry to the bottle config" genuinely works — for
-quay, MCR, registry.k8s.io, or any unauthenticated internal registry.
-Docker Hub and GHCR are the ones that need the strip resolved. The
-acceptance test uses quay for exactly this reason.
-
-Resolving it for Docker Hub means picking one of:
-
-1. **Per-route opt-in to preserve client Authorization.** Smallest
-   change. Note the compounding effect on exactly these routes: the DLP
-   scoping above already removed `token_patterns` there, so a
-   preserved-auth registry route is one where the agent may send bearer
-   tokens that neither the strip nor the pattern detector inspects.
-   `known_secrets` still applies, so the bottle's real credentials are
-   still caught.
-2. **A registry-aware gateway** that performs the token dance itself and
-   injects the result. Preserves the invariant fully; materially more
-   work, and it makes the gateway speak a specific registry protocol.
-3. **Pre-seed images at provision time** (host-side `container image
-   save` into podman storage), so bottles never pull at runtime.
-   Preserves the invariant, and limits nested containers to
-   pre-approved images — which fits the custody positioning, at the cost
-   of no ad-hoc `docker pull`.
-4. **Stop.** Nested containers are not supported on this backend.
-
-### Podman 4.3.1 silently swallows container exit codes
-
-Debian bookworm — which the current agent base image is built on —
-ships podman 4.3.1. Through its Docker-compatible API, `docker run`
-returns 0 no matter what the container did:
-
-| Command | podman 4.3.1 | podman 5.4.2 |
-| --- | --- | --- |
-| `docker run … sh -c 'exit 7'` (compat API) | **0** | 7 |
-| `docker run … sh -c 'exit 0'` (compat API) | 0 | 0 |
-| `podman run … sh -c 'exit 7'` (native) | 7 | 7 |
-
-This is worse than a broken feature: every failing command an agent runs
-via `docker run` reports success. A test suite, a build step, or a CI
-script inside a bottle would pass while failing. It also silently
-defeated the acceptance test's egress-containment assertion, which is
-why that assertion now checks an in-band marker rather than an exit
-code.
-
-Podman 5.4.2 (Debian trixie) fixes it, but needs two packages that
-bookworm's podman does not: `passt` (podman 5's default network tool)
-and `nftables` (netavark shells out to `nft`; without it every run fails
-with `unable to upgrade to tcp, received 500`). With both installed,
-exit codes propagate correctly and the compat API behaves.
-
-The open question this leaves is where podman 5 comes from, since the
-agent base is bookworm-based:
-
-1. **Move the agent images to Debian trixie.** Trixie is current stable.
-   Correct, and the blast radius is every bottle, not just this feature.
-2. **Drop the compat socket and use podman natively** (`podman-docker`
-   provides a `docker` shim; compose comes from `podman-compose`).
-   Native podman propagates exit codes correctly even on 4.3.1. Contained
-   to this feature, at the cost of `docker compose` becoming
-   `docker-compose`/`podman-compose`.
-3. **Ship bookworm's podman 4.3.1 with the compat socket** — not viable.
-   Silent false success is a correctness bug agents cannot see.
-
-## Recommendation
-
-1. Do not revive rootless Docker on this backend. This document is the
-   record of why.
-2. Nested containers, if wanted, come from podman under the
-   single-mapping constraint — with the explicit understanding that the
-   nested-container boundary carries no security weight. `root` in a
-   nested container is the agent user outside it.
-3. Nested containers are therefore a build/test convenience. The bottle
-   remains the security boundary, exactly as it was.

scripts/coverage.sh

@@ -1,15 +1,19 @@
 #!/usr/bin/env bash
 # Combined unit + integration coverage (see docs/decisions/0004-coverage-policy.md).
 #
-# Runs the unit suite, then appends the integration suite (which skips
-# cleanly when Docker / the backend CLIs are unavailable), and prints one
-# combined report. The integration suite is what scores the subprocess /
-# backend orchestration modules, so the number here is the policy's
-# yardstick — not the unit-only badge.
+# Two modes:
 #
-# Usage:
-#   scripts/coverage.sh            # combined report
-#   scripts/coverage.sh critical   # also report just the critical modules
+#   scripts/coverage.sh [critical]
+#     Run mode (default, for local dev): executes the unit suite then the
+#     integration suite under coverage and prints a combined report.
+#
+#   scripts/coverage.sh aggregate [critical]
+#     Aggregate mode (used by CI): combines pre-existing .coverage.* files
+#     produced by individual test jobs and prints a combined report. No tests
+#     are re-executed; no KVM or Docker dependency.
+#
+# Pass "critical" as the last argument in either mode to also report just the
+# critical modules (ADR 0004 target: 90%).
 set -euo pipefail
 
 cd "$(dirname "$0")/.."
@@ -21,6 +25,22 @@ PY="${PYTHON:-python3}"
 # README "core coverage" badge can't drift; comma-join it for --include.
 CRITICAL=$(grep -vE '^[[:space:]]*(#|$)' scripts/critical-modules.txt | paste -sd, -)
 
+if [ "${1:-}" = "aggregate" ]; then
+    # Aggregate mode: combine .coverage.* artifacts already in the workspace.
+    echo "== combining coverage artifacts ==" >&2
+    "$PY" -m coverage combine
+
+    echo "== combined report ==" >&2
+    "$PY" -m coverage report -m
+
+    if [ "${2:-}" = "critical" ]; then
+        echo "== critical modules (ADR 0004 target: 90%) ==" >&2
+        "$PY" -m coverage report --include="$CRITICAL"
+    fi
+    exit 0
+fi
+
+# Run mode (default): execute both suites under coverage in this process.
 rm -f .coverage
 
 echo "== unit ==" >&2

tests/integration/test_macos_rootless_podman_spike.py

@@ -1,154 +0,0 @@
-"""Live-Mac acceptance spike for guest-local rootless podman (issue #392).
-
-Run explicitly on an Apple Silicon/macOS 26 host:
-
-    BOT_BOTTLE_ROOTLESS_PODMAN_SPIKE=1 \
-      python3 -m unittest tests.integration.test_macos_rootless_podman_spike -v
-
-The opt-in is deliberate: ordinary Linux CI cannot execute Apple Container.
-
-Podman rather than Docker because Apple Container's capability bounding set
-omits CAP_SYS_ADMIN; see
-docs/research/rootless-docker-in-apple-container-spike.md.  The agent-facing
-surface is still `docker` and `docker compose`, which talk to podman's
-Docker-compatible API socket.
-"""
-
-from __future__ import annotations
-
-import os
-import platform
-import shutil
-import tempfile
-import unittest
-from pathlib import Path
-
-from bot_bottle.backend import BottleSpec, get_bottle_backend
-from bot_bottle.manifest import ManifestIndex
-
-
-@unittest.skipUnless(
-    platform.system() == "Darwin"
-    and os.environ.get("BOT_BOTTLE_ROOTLESS_PODMAN_SPIKE") == "1",
-    "requires an explicit live-Mac rootless-podman spike run",
-)
-class TestMacosRootlessPodmanSpike(unittest.TestCase):
-    def test_compose_stays_inside_registered_bottle(self) -> None:
-        workspace = Path(tempfile.mkdtemp(prefix="rootless-podman-spike."))
-        stage = Path(tempfile.mkdtemp(prefix="rootless-podman-stage."))
-        try:
-            (workspace / "index.html").write_text("bottle-compose-ok\n")
-            (workspace / "compose.yaml").write_text(
-                "services:\n"
-                "  web:\n"
-                "    image: quay.io/prometheus/busybox\n"
-                "    working_dir: /workspace\n"
-                "    command: httpd -f -p 8000 -h /workspace\n"
-                "    volumes: ['.:/workspace']\n"
-                "    ports: ['18080:8000']\n",
-                encoding="utf-8",
-            )
-            manifest = ManifestIndex.from_json_obj({
-                "bottles": {"dev": {
-                    "docker_access": True,
-                    # A deliberately tiny image. Pulling a ~165MB one
-                    # OOM-kills the shared egress proxy, which buffers whole
-                    # response bodies to scan them — a real defect, but a
-                    # separate one from what this test covers. See the
-                    # research note.
-                    #
-                    # quay.io deliberately, not Docker Hub: the egress proxy
-                    # strips agent-set Authorization (so an agent cannot
-                    # smuggle a credential out in a header), and Docker Hub
-                    # requires a client-fetched, per-scope bearer token that
-                    # the strip therefore removes. quay serves manifests with
-                    # no Authorization at all, so a plain route is enough.
-                    #
-                    # token_patterns is still scoped off: registry traffic
-                    # carries bearer JWTs by protocol and trips the generic
-                    # rule. known_secrets stays on — it matches the bottle's
-                    # own credentials, which is the detector that catches
-                    # real exfil.
-                    "egress": {"routes": [
-                        {"host": "quay.io", "dlp": {
-                            "outbound_detectors": ["known_secrets"],
-                        }},
-                        {"host": "cdn01.quay.io", "dlp": {
-                            "outbound_detectors": ["known_secrets"],
-                        }},
-                    ]},
-                }},
-                "agents": {"spike": {
-                    "bottle": "dev", "skills": [], "prompt": "",
-                }},
-            })
-            spec = BottleSpec(
-                manifest=manifest,
-                agent_name="spike",
-                copy_cwd=True,
-                user_cwd=str(workspace),
-            )
-            backend = get_bottle_backend("macos-container")
-            plan = backend.prepare(spec, stage_dir=stage)
-            with backend.launch(plan) as bottle:
-                workdir = plan.workspace_plan.workdir
-                checks = (
-                    "docker info >/dev/null && docker compose version && "
-                    f"cd {workdir} && docker compose up -d --wait && "
-                    "curl --fail --silent http://127.0.0.1:18080/ | "
-                    "grep -q bottle-compose-ok"
-                )
-                result = bottle.exec(checks)
-                self.assertEqual(
-                    0, result.returncode,
-                    f"stdout={result.stdout!r}\nstderr={result.stderr!r}",
-                )
-                # podman's compat API reports rootlessness through its own
-                # native endpoint; the Docker-shaped SecurityOptions field does
-                # not carry it.
-                inspect = bottle.exec(
-                    "podman info --format '{{.Host.Security.Rootless}}'"
-                )
-                self.assertIn("true", inspect.stdout.lower())
-                self.assertEqual(
-                    0,
-                    bottle.exec(
-                        "test \"$(id -u)\" -ne 0"
-                    ).returncode,
-                    "the podman service must not be running as bottle root",
-                )
-                self.assertNotEqual(
-                    0,
-                    bottle.exec("test -S /var/run/docker.sock").returncode,
-                    "spike must never expose a host/rootful Docker socket",
-                )
-                # Asserted on an in-band marker, not on `docker run`'s exit
-                # code: podman 4.3.1's Docker-compat API swallows the
-                # container's status and returns 0 for everything, so an
-                # exit-code assertion here passes whether egress was blocked
-                # or wide open. That silent false pass is worse than no check
-                # at all, and it is exactly this check — the one proving a
-                # nested container cannot escape the egress path.
-                #
-                # busybox ships wget, so a failure here means egress was
-                # refused rather than the binary being absent.
-                direct = bottle.exec(
-                    "docker run --rm --env HTTP_PROXY= --env HTTPS_PROXY= "
-                    "--env http_proxy= --env https_proxy= "
-                    "quay.io/prometheus/busybox sh -c "
-                    "'wget -T 4 -qO- https://evil.example.com/ "
-                    "&& echo ESCAPED || echo CONTAINED'"
-                )
-                self.assertIn(
-                    "CONTAINED", direct.stdout,
-                    "an inner container obtained direct, unproxied egress: "
-                    f"stdout={direct.stdout!r} stderr={direct.stderr!r}",
-                )
-                self.assertNotIn("ESCAPED", direct.stdout)
-        finally:
-            shutil.rmtree(workspace, ignore_errors=True)
-            shutil.rmtree(stage, ignore_errors=True)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/integration/test_orchestrator_docker_control_plane_auth.py

@@ -34,6 +34,7 @@ from tests._docker import skip_unless_docker
 # image instead of leaking a new dangling tag on every invocation.
 _TEST_ORCHESTRATOR_IMAGE = "bot-bottle-orchestrator:itest"
 _TEST_GATEWAY_IMAGE = "bot-bottle-gateway:itest"
+_TEST_INFRA_IMAGE = "bot-bottle-infra:itest"
 
 
 @skip_unless_docker()
@@ -69,20 +70,17 @@ class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
         os.environ["BOT_BOTTLE_ROOT"] = cls._tmp.name
         cls.addClassCleanup(_restore_root)
 
-        orchestrator_name = f"bot-bottle-orch-itest-{suffix}"
-        gateway_name = f"bot-bottle-gw-itest-{suffix}"
+        infra_name = f"bot-bottle-infra-itest-{suffix}"
         network = f"bot-bottle-net-itest-{suffix}"
         host_root = Path(cls._tmp.name)
         cls.addClassCleanup(
-            cls._teardown_docker, orchestrator_name, gateway_name, network, host_root
+            cls._teardown_docker, infra_name, network, host_root
         )
 
         cls.svc = OrchestratorService(
-            orchestrator_name=orchestrator_name,
-            gateway_name=gateway_name,
+            infra_name=infra_name,
             network=network,
-            image=_TEST_ORCHESTRATOR_IMAGE,
-            gateway_image=_TEST_GATEWAY_IMAGE,
+            image=_TEST_INFRA_IMAGE,
             port=20000 + secrets.randbelow(10000),
             host_root=host_root,
         )
@@ -91,23 +89,23 @@ class TestDockerControlPlaneAuthIntegration(unittest.TestCase):
 
     @staticmethod
     def _teardown_docker(
-        orchestrator_name: str, gateway_name: str, network: str, host_root: Path
+        infra_name: str, network: str, host_root: Path
     ) -> None:
         subprocess.run(
-            ["docker", "rm", "--force", orchestrator_name, gateway_name],
+            ["docker", "rm", "--force", infra_name],
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
         )
         subprocess.run(
             ["docker", "network", "rm", network],
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
         )
-        # The orchestrator container (no USER directive) wrote the registry
+        # The infra container (no USER directive) wrote the registry
         # DB as root into the throwaway host_root; chown it back so the
         # (non-root) tempdir cleanup can remove it. Same workaround
         # test_multitenant_isolation.py uses for the identical bind mount.
         subprocess.run(
             ["docker", "run", "--rm", "-v", f"{host_root}:/r",
-             "--entrypoint", "chown", _TEST_GATEWAY_IMAGE, "-R",
+             "--entrypoint", "chown", _TEST_INFRA_IMAGE, "-R",
              f"{os.getuid()}:{os.getgid()}", "/r"],
             stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL, check=False,
         )

tests/integration/test_sandbox_escape.py

@@ -172,7 +172,7 @@ class TestSandboxEscape(unittest.TestCase):
         # base image without producing five confusing
         # command-not-found failures down the suite.
         missing: list[str] = []
-        for tool in ("curl", "git", "dig"):
+        for tool in ("curl", "git", "dig", "ssh"):
             r = cls._bottle.exec(f"command -v {tool} >/dev/null 2>&1")
             if r.returncode != 0:
                 missing.append(tool)

tests/unit/test_agent_provider.py

@@ -9,11 +9,15 @@ import unittest
 from pathlib import Path
 
 from bot_bottle.agent_provider import (
+    CLAUDE_HOST_CREDENTIAL_HOSTS,
     CODEX_HOST_CREDENTIAL_HOSTS,
     build_agent_provision_plan,
     prompt_args,
 )
-from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
+from bot_bottle.egress import (
+    CLAUDE_HOST_CREDENTIAL_TOKEN_REF,
+    CODEX_HOST_CREDENTIAL_TOKEN_REF,
+)
 
 
 def _jwt(exp: int) -> str:
@@ -292,6 +296,67 @@ class TestAgentProviderRuntime(unittest.TestCase):
             )
         self.assertEqual({}, plan.provisioned_env)
 
+    def test_claude_forward_host_credentials_populates_egress_route(self):
+        access_token = "sk-ant-oat01-test-key"  # gitleaks:allow
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            home = Path(tmp) / "host-claude"
+            cred_dir = home / ".claude"
+            cred_dir.mkdir(parents=True)
+            (cred_dir / ".credentials.json").write_text(json.dumps({
+                "claudeAiOauth": {"accessToken": access_token},
+            }))
+            plan = build_agent_provision_plan(
+                template="claude",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                forward_host_credentials=True,
+                host_env={"HOME": str(home)},
+            )
+        self.assertEqual(1, len(plan.egress_routes))
+        route = plan.egress_routes[0]
+        self.assertIn(route.host, CLAUDE_HOST_CREDENTIAL_HOSTS)
+        self.assertEqual("Bearer", route.auth_scheme)
+        self.assertEqual(CLAUDE_HOST_CREDENTIAL_TOKEN_REF, route.token_ref)
+        self.assertEqual("egress-placeholder", plan.env_vars["CLAUDE_CODE_OAUTH_TOKEN"])
+        self.assertEqual(frozenset({"CLAUDE_CODE_OAUTH_TOKEN"}), plan.hidden_env_names)
+
+    def test_claude_forward_host_credentials_populates_provisioned_env(self):
+        access_token = "sk-ant-oat01-test-key"  # gitleaks:allow
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            home = Path(tmp) / "host-claude"
+            cred_dir = home / ".claude"
+            cred_dir.mkdir(parents=True)
+            (cred_dir / ".credentials.json").write_text(json.dumps({
+                "claudeAiOauth": {"accessToken": access_token},
+            }))
+            plan = build_agent_provision_plan(
+                template="claude",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                forward_host_credentials=True,
+                host_env={"HOME": str(home)},
+            )
+        self.assertEqual(
+            {CLAUDE_HOST_CREDENTIAL_TOKEN_REF: access_token},
+            plan.provisioned_env,
+        )
+
+    def test_claude_without_forward_host_credentials_has_empty_provisioned_env(self):
+        with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
+            plan = build_agent_provision_plan(
+                template="claude",
+                dockerfile="",
+                state_dir=Path(tmp),
+                instance_name="bot-bottle-test",
+                prompt_file=Path(tmp) / "prompt.txt",
+                forward_host_credentials=False,
+            )
+        self.assertEqual({}, plan.provisioned_env)
+
     def test_pi_plan_writes_default_ollama_models(self):
         with tempfile.TemporaryDirectory(prefix="bb-provider.") as tmp:
             plan = build_agent_provision_plan(

tests/unit/test_backend_util.py

@@ -0,0 +1,29 @@
+"""Unit: shared cross-backend helpers in backend/util.py."""
+
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from bot_bottle.backend import util as backend_util
+
+
+class TestPollCaCert(unittest.TestCase):
+    def test_returns_pem_on_first_success(self) -> None:
+        result = backend_util.poll_ca_cert(lambda: "PEM", timeout=1.0)
+        self.assertEqual("PEM", result)
+
+    def test_raises_timeout_error_when_cert_never_appears(self) -> None:
+        with self.assertRaises(TimeoutError):
+            backend_util.poll_ca_cert(lambda: None, timeout=0.0)
+
+    def test_polls_until_cert_appears(self) -> None:
+        responses = iter([None, None, "-----BEGIN CERTIFICATE-----\n"])
+        with patch("bot_bottle.backend.util.time.sleep") as mock_sleep:
+            result = backend_util.poll_ca_cert(lambda: next(responses), timeout=5.0)
+        self.assertTrue(result.startswith("-----BEGIN CERTIFICATE-----"))
+        self.assertEqual(2, mock_sleep.call_count)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_builtin_agent_images.py

@@ -0,0 +1,49 @@
+"""Unit contracts shared by all built-in agent images."""
+
+from __future__ import annotations
+
+import re
+import unittest
+from pathlib import Path
+
+
+_CONTRIB_DIR = Path(__file__).resolve().parents[2] / "bot_bottle/contrib"
+_AGENT_DOCKERFILES = tuple(sorted(_CONTRIB_DIR.glob("*/Dockerfile")))
+
+
+class TestBuiltinAgentImages(unittest.TestCase):
+    def test_all_use_debian_trixie_stable(self):
+        self.assertTrue(_AGENT_DOCKERFILES)
+        for dockerfile in _AGENT_DOCKERFILES:
+            with self.subTest(provider=dockerfile.parent.name):
+                self.assertRegex(
+                    dockerfile.read_text(),
+                    r"(?m)^FROM node:22-trixie-slim\s*$",
+                )
+
+    def test_all_install_podman(self):
+        for dockerfile in _AGENT_DOCKERFILES:
+            with self.subTest(provider=dockerfile.parent.name):
+                self.assertRegex(
+                    dockerfile.read_text(),
+                    re.compile(r"(?m)^\s*podman(?:\s|\\|$)"),
+                )
+
+    def test_all_install_ssh_client(self):
+        for dockerfile in _AGENT_DOCKERFILES:
+            with self.subTest(provider=dockerfile.parent.name):
+                self.assertRegex(
+                    dockerfile.read_text(),
+                    re.compile(r"(?m)^\s*openssh-client(?:\s|\\|$)"),
+                )
+
+    def test_all_prepare_node_git_config_directory(self):
+        for dockerfile in _AGENT_DOCKERFILES:
+            with self.subTest(provider=dockerfile.parent.name):
+                dockerfile_text = dockerfile.read_text()
+                self.assertIn("install -d -o node -g node", dockerfile_text)
+                self.assertIn("/home/node/.config/git", dockerfile_text)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_cli_login.py

@@ -0,0 +1,268 @@
+"""Unit tests for bb login command."""
+
+from __future__ import annotations
+
+import json
+import os
+import tempfile
+import unittest
+import urllib.error
+from email.message import Message
+from typing import Any
+from unittest.mock import MagicMock, patch
+
+
+class TestFlagParsing(unittest.TestCase):
+    def test_console_url_flag(self) -> None:
+        from bot_bottle.cli.login import _flag
+        self.assertEqual(_flag(["--console-url", "http://x"], "--console-url"), "http://x")
+
+    def test_console_url_equals_form(self) -> None:
+        from bot_bottle.cli.login import _flag
+        self.assertEqual(
+            _flag(["--console-url=http://x"], "--console-url"), "http://x"
+        )
+
+    def test_label_flag(self) -> None:
+        from bot_bottle.cli.login import _flag
+        self.assertEqual(_flag(["--label", "my-mac"], "--label"), "my-mac")
+
+    def test_missing_flag_returns_none(self) -> None:
+        from bot_bottle.cli.login import _flag
+        self.assertIsNone(_flag([], "--console-url"))
+
+
+class TestHttpHelpers(unittest.TestCase):
+    def test_post_sends_json_and_decodes_response(self) -> None:
+        from bot_bottle.cli.login import _post
+
+        response = MagicMock()
+        response.__enter__.return_value.read.return_value = b'{"ok": true}'
+        with patch("urllib.request.urlopen", return_value=response) as urlopen:
+            self.assertEqual(
+                _post("http://console/start", {"label": "host"}), {"ok": True}
+            )
+
+        request = urlopen.call_args.args[0]
+        self.assertEqual(request.data, b'{"label": "host"}')
+        self.assertEqual(request.get_header("Content-type"), "application/json")
+
+    def test_get_decodes_success_response(self) -> None:
+        from bot_bottle.cli.login import _get
+
+        response = MagicMock()
+        response.__enter__.return_value.status = 200
+        response.__enter__.return_value.read.return_value = b'{"status": "pending"}'
+        with patch("urllib.request.urlopen", return_value=response):
+            self.assertEqual(
+                _get("http://console/status"), (200, {"status": "pending"})
+            )
+
+    def test_get_returns_http_error_status(self) -> None:
+        from bot_bottle.cli.login import _get
+
+        error = urllib.error.HTTPError(
+            "http://console/status", 410, "gone", Message(), None
+        )
+        with patch("urllib.request.urlopen", side_effect=error):
+            self.assertEqual(_get("http://console/status"), (410, {}))
+
+
+class TestSaveCredentials(unittest.TestCase):
+    def test_writes_json_and_sets_perms(self) -> None:
+        from bot_bottle.cli.login import _save_credentials
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+                path = _save_credentials("http://c", "hid", "at", "rt")
+            self.assertTrue(path.exists())
+            data = json.loads(path.read_text())
+            self.assertEqual(data["url"], "http://c")
+            self.assertEqual(data["host_id"], "hid")
+            self.assertEqual(data["access_token"], "at")
+            self.assertEqual(data["refresh_token"], "rt")
+            self.assertEqual(oct(path.stat().st_mode & 0o777), oct(0o600))
+
+    def test_temp_file_is_private_before_replace(self) -> None:
+        """Temp file must be 0600 at the moment os.replace is called."""
+        from bot_bottle.cli.login import _save_credentials
+        from pathlib import Path as _Path
+
+        tmp_perms_at_replace: list[int] = []
+        real_replace = os.replace
+
+        def _spy_replace(
+            src: str | os.PathLike[str], dst: str | os.PathLike[str]
+        ) -> None:
+            tmp_perms_at_replace.append(_Path(src).stat().st_mode & 0o777)
+            real_replace(src, dst)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+                with patch("os.replace", side_effect=_spy_replace):
+                    path = _save_credentials("http://c", "hid", "at", "rt")
+                self.assertEqual(len(tmp_perms_at_replace), 1)
+                self.assertEqual(oct(tmp_perms_at_replace[0]), oct(0o600))
+                self.assertEqual(oct(path.stat().st_mode & 0o777), oct(0o600))
+
+    def test_cleanup_on_write_failure(self) -> None:
+        """Temp file is removed and no credentials remain if replace fails."""
+        from bot_bottle.cli.login import _save_credentials
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+                with patch("os.replace", side_effect=OSError("disk full")):
+                    with self.assertRaises(OSError):
+                        _save_credentials("http://c", "hid", "at", "rt")
+            leftovers = [f for f in os.listdir(tmp) if f.startswith(".console-")]
+            self.assertEqual(leftovers, [])
+
+
+class TestCmdLoginMissingUrl(unittest.TestCase):
+    def test_help_returns_0(self) -> None:
+        from bot_bottle.cli.login import cmd_login
+
+        self.assertEqual(cmd_login(["--help"]), 0)
+
+    def test_returns_1_without_url(self) -> None:
+        from bot_bottle.cli.login import cmd_login
+        with patch.dict(os.environ, {}, clear=True):
+            os.environ.pop("BB_CONSOLE_URL", None)
+            result = cmd_login([])
+        self.assertEqual(result, 1)
+
+    def test_reads_env_var(self) -> None:
+        """Exits 1 (network error) not because of missing URL when env var is set."""
+        from bot_bottle.cli.login import cmd_login
+
+        def _fail_post(_url: str, _payload: dict[str, Any]) -> dict[str, Any]:
+            raise OSError("connection refused")
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(
+                os.environ,
+                {
+                    "BB_CONSOLE_URL": "http://localhost:9999",
+                    "BOT_BOTTLE_ROOT": tmp,
+                },
+            ):
+                with patch("bot_bottle.cli.login._post", side_effect=_fail_post):
+                    result = cmd_login([])
+        self.assertEqual(result, 1)
+
+
+class TestCmdLoginFlow(unittest.TestCase):
+    def _run_with_mocks(
+        self, poll_responses: list[dict[str, Any]], tmp: str
+    ) -> int:
+        from bot_bottle.cli.login import cmd_login
+
+        start_resp = {
+            "device_code": "dc123",
+            "user_code": "ABC-DEF",
+            "expires_in": 300,
+            "poll_interval": 0,
+        }
+
+        poll_iter = iter(poll_responses)
+
+        def _fake_post(
+            _url: str, _payload: dict[str, Any]
+        ) -> dict[str, Any]:
+            return start_resp
+
+        def _fake_get(_url: str) -> tuple[int, dict[str, Any]]:
+            return 200, next(poll_iter, {"status": "pending"})
+
+        with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+            with patch("bot_bottle.cli.login._post", side_effect=_fake_post):
+                with patch("bot_bottle.cli.login._get", side_effect=_fake_get):
+                    with patch("time.sleep"):
+                        return cmd_login(["--console-url", "http://console"])
+
+    def test_approved_flow_returns_0(self) -> None:
+        approved = {
+            "status": "approved",
+            "host_id": "hid",
+            "access_token": "at",
+            "refresh_token": "rt",
+        }
+        with tempfile.TemporaryDirectory() as tmp:
+            result = self._run_with_mocks(
+                [{"status": "pending"}, approved], tmp
+            )
+            self.assertEqual(result, 0)
+            with open(os.path.join(tmp, "console.json"), encoding="utf-8") as f:
+                creds = json.loads(f.read())
+            self.assertEqual(creds["host_id"], "hid")
+
+    def test_denied_flow_returns_1(self) -> None:
+        with tempfile.TemporaryDirectory() as tmp:
+            result = self._run_with_mocks([{"status": "denied"}], tmp)
+        self.assertEqual(result, 1)
+
+    def test_timeout_returns_1(self) -> None:
+        from bot_bottle.cli.login import cmd_login
+
+        start_resp = {
+            "device_code": "dc",
+            "user_code": "ZZZ-ZZZ",
+            "expires_in": 0,  # already expired; loop never runs
+            "poll_interval": 2,
+        }
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+                with patch("bot_bottle.cli.login._post", return_value=start_resp):
+                    result = cmd_login(["--console-url", "http://console"])
+        self.assertEqual(result, 1)
+
+    def test_poll_interval_from_server_is_used(self) -> None:
+        """time.sleep must be called with the server-provided poll_interval."""
+        from bot_bottle.cli.login import cmd_login
+
+        server_interval = 7
+        start_resp = {
+            "device_code": "dc",
+            "user_code": "ABC-DEF",
+            "expires_in": 300,
+            "poll_interval": server_interval,
+        }
+        approved = {
+            "status": "approved",
+            "host_id": "hid",
+            "access_token": "at",
+            "refresh_token": "rt",
+        }
+        poll_iter = iter([{"status": "pending"}, approved])
+
+        def _fake_get(_url: str) -> tuple[int, dict[str, str]]:
+            return 200, next(poll_iter)
+
+        with tempfile.TemporaryDirectory() as tmp:
+            with patch.dict(os.environ, {"BOT_BOTTLE_ROOT": tmp}):
+                with patch("bot_bottle.cli.login._post", return_value=start_resp):
+                    with patch(
+                        "bot_bottle.cli.login._get", side_effect=_fake_get
+                    ):
+                        with patch("time.sleep") as mock_sleep:
+                            result = cmd_login(["--console-url", "http://console"])
+
+        self.assertEqual(result, 0)
+        self.assertTrue(mock_sleep.called)
+        for call in mock_sleep.call_args_list:
+            self.assertEqual(call.args[0], server_interval)
+
+
+class TestDispatcherRegistration(unittest.TestCase):
+    def test_login_in_commands(self) -> None:
+        from bot_bottle.cli import COMMANDS
+        self.assertIn("login", COMMANDS)
+
+    def test_login_in_no_migration(self) -> None:
+        from bot_bottle.cli import NO_MIGRATION_COMMANDS
+        self.assertIn("login", NO_MIGRATION_COMMANDS)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_cli_module_entry.py

@@ -0,0 +1,40 @@
+"""The CLI package is runnable as `python -m bot_bottle.cli`."""
+
+from __future__ import annotations
+
+import subprocess
+import sys
+import unittest
+from pathlib import Path
+
+_REPO_ROOT = Path(__file__).resolve().parents[2]
+
+
+def _run(*args: str) -> subprocess.CompletedProcess[str]:
+    return subprocess.run(
+        [sys.executable, "-m", "bot_bottle.cli", *args],
+        cwd=_REPO_ROOT,
+        capture_output=True,
+        text=True,
+        check=False,
+    )
+
+
+class TestModuleEntry(unittest.TestCase):
+    def test_help_exits_zero(self) -> None:
+        result = _run("--help")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("login", result.stderr)
+
+    def test_no_args_prints_usage(self) -> None:
+        # main() returns 2 with no command, matching the cli.py entry point.
+        self.assertEqual(_run().returncode, 2)
+
+    def test_subcommand_help_reaches_handler(self) -> None:
+        result = _run("login", "--help")
+        self.assertEqual(result.returncode, 0, result.stderr)
+        self.assertIn("--console-url", result.stderr)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_cli_start_headless.py

@@ -176,6 +176,18 @@ class TestCmdStartHeadless(unittest.TestCase):
         self.assertEqual("researcher-2", self._spec().label)
 
 
+    def test_cached_images_sets_cached_policy(self):
+        start_mod.cmd_start(
+            ["--headless", "--cached-images", "researcher", "--bottle", "claude",
+             "--prompt", "Do it"]
+        )
+        self.assertEqual("cached", self._spec().image_policy)
+
+    def test_cached_images_requires_headless(self):
+        with self.assertRaises(Die):
+            start_mod.cmd_start(["--cached-images", "researcher"])
+        self._launch_mock.assert_not_called()
+
 
 class TestPrepareWithPreflight(unittest.TestCase):
     """prepare_with_preflight calls render_preflight with the plan and backend name."""

tests/unit/test_cli_start_selector.py

@@ -65,6 +65,12 @@ class TestCmdStartSelector(unittest.TestCase):
         )
         self._modal_patch.start()
 
+        self._image_policy_patch = patch(
+            "bot_bottle.cli.start._select_image_policy",
+            return_value="fresh",
+        )
+        self._image_policy_patch.start()
+
         self._env_patch = patch.dict(os.environ, {}, clear=False)
         self._env_patch.start()
         os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -75,6 +81,7 @@ class TestCmdStartSelector(unittest.TestCase):
         self._agent_picker_patch.stop()
         self._bottle_picker_patch.stop()
         self._modal_patch.stop()
+        self._image_policy_patch.stop()
         self._env_patch.stop()
 
     # ------------------------------------------------------------------
@@ -133,6 +140,19 @@ class TestCmdStartSelector(unittest.TestCase):
         spec = self._launch_mock.call_args[0][0]
         self.assertEqual(("claude", "dev"), spec.bottle_names)
 
+    def test_image_policy_forwarded_to_spec(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value="cached"):
+            start_mod.cmd_start(["researcher"])
+        self._launch_mock.assert_called_once()
+        spec = self._launch_mock.call_args[0][0]
+        self.assertEqual("cached", spec.image_policy)
+
+    def test_image_policy_cancel_returns_0(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value=None):
+            rc = start_mod.cmd_start(["researcher"])
+        self.assertEqual(0, rc)
+        self._launch_mock.assert_not_called()
+
     def test_empty_bottle_selection_forwarded(self):
         self._bottle_picker_mock.return_value = []
         start_mod.cmd_start(["researcher"])
@@ -214,6 +234,7 @@ class TestCmdStartLabelCollision(unittest.TestCase):
         ).start()
         # Stub the bottle picker to always return a selection.
         patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start()
+        patch("bot_bottle.cli.start._select_image_policy", return_value="fresh").start()
         self.addCleanup(patch.stopall)
 
     def test_no_collision_proceeds_without_reprompt(self):

tests/unit/test_cli_start_stale.py

@@ -0,0 +1,146 @@
+"""Unit: _launch_bottle StaleImageError handling.
+
+Exercises prelaunch_checks / backend.launch flow:
+  - headless mode → die on stale
+  - interactive mode, user declines → stop without launching
+  - interactive mode, user confirms → skip stale check and launch once
+"""
+
+from __future__ import annotations
+
+import io
+import tempfile
+import unittest
+from types import SimpleNamespace
+from typing import Any, cast
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.image_cache import StaleImageError
+from bot_bottle.log import Die
+
+
+def _fake_plan() -> Any:
+    provision = SimpleNamespace(startup_args=())
+    return cast(Any, SimpleNamespace(
+        agent_provision=provision,
+        agent_provider_template="claude",
+        slug="dev-abc",
+    ))
+
+
+def _ok_cm(bottle: Any) -> MagicMock:
+    """Return a context-manager mock that yields `bottle`."""
+    cm = MagicMock()
+    cm.__enter__ = MagicMock(return_value=bottle)
+    cm.__exit__ = MagicMock(return_value=False)
+    return cm
+
+
+class TestLaunchBottleStaleHandling(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.")
+
+    def _spec(self) -> Any:
+        from bot_bottle.backend import BottleSpec
+        from bot_bottle.manifest import ManifestIndex
+        idx = ManifestIndex.from_json_obj({
+            "bottles": {"dev": {}},
+            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
+        })
+        return BottleSpec(
+            manifest=idx,
+            agent_name="demo",
+            copy_cwd=False,
+            user_cwd=self._tmp,
+            identity="dev-abc",
+        )
+
+    def _run_launch(self, **patch_kwargs: Any) -> int:
+        import bot_bottle.cli.start as start_mod
+        spec = self._spec()
+        with patch.object(start_mod, "prepare_with_preflight",
+                          return_value=(_fake_plan(), "dev-abc")), \
+             patch.object(start_mod, "settle_state"), \
+             patch.object(start_mod, "info"):
+            return start_mod._launch_bottle(
+                spec,
+                dry_run=False,
+                backend_name="docker",
+                **patch_kwargs,
+            )
+
+    def test_headless_stale_calls_die(self) -> None:
+        """In headless mode (assume_yes=True), a StaleImageError from prelaunch_checks must call die()."""
+        import bot_bottle.cli.start as start_mod
+
+        backend_mock = MagicMock()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+
+        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
+             patch.object(start_mod, "die", side_effect=Die()):
+            with self.assertRaises(Die):
+                self._run_launch(assume_yes=True)
+
+        backend_mock.launch.assert_not_called()
+
+    def test_interactive_user_declines_stops_before_launch(self) -> None:
+        """Interactive user answering 'n' → launch is never called."""
+        import bot_bottle.cli.start as start_mod
+
+        backend_mock = MagicMock()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+
+        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
+             patch.object(start_mod, "read_tty_line", return_value="n"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        backend_mock.launch.assert_not_called()
+
+    def test_interactive_user_confirms_launches_once(self) -> None:
+        """Interactive user answering 'y' → prelaunch stale error is bypassed; launch called once."""
+        import bot_bottle.cli.start as start_mod
+
+        bottle_mock = MagicMock()
+        bottle_mock.name = "dev-abc"
+
+        backend_mock = MagicMock()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+        backend_mock.launch.return_value = _ok_cm(bottle_mock)
+
+        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
+             patch.object(start_mod, "read_tty_line", return_value="y"), \
+             patch.object(start_mod, "attach_agent", return_value=0), \
+             patch.object(start_mod, "capture_claude_session_state"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        backend_mock.prelaunch_checks.assert_called_once()
+        backend_mock.launch.assert_called_once()
+
+    def test_interactive_yes_uppercase_also_accepted(self) -> None:
+        """'Y' or 'YES' should also be accepted as confirmation."""
+        import bot_bottle.cli.start as start_mod
+
+        bottle_mock = MagicMock()
+        bottle_mock.name = "dev-abc"
+
+        backend_mock = MagicMock()
+        backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
+        backend_mock.launch.return_value = _ok_cm(bottle_mock)
+
+        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
+             patch.object(start_mod, "read_tty_line", return_value="YES"), \
+             patch.object(start_mod, "attach_agent", return_value=0), \
+             patch.object(start_mod, "capture_claude_session_state"), \
+             patch("sys.stderr", new_callable=io.StringIO):
+            rc = self._run_launch(assume_yes=False)
+
+        self.assertEqual(0, rc)
+        backend_mock.launch.assert_called_once()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_config_store.py

@@ -0,0 +1,86 @@
+"""Unit tests for the host-side configuration store."""
+
+from __future__ import annotations
+
+import sqlite3
+import tempfile
+import unittest
+from pathlib import Path
+
+from bot_bottle.config_store import (
+    DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+    ConfigStore,
+)
+from bot_bottle.store_manager import StoreManager
+
+
+class TestConfigStore(unittest.TestCase):
+    def test_cached_image_warning_days_defaults_to_one(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+    def test_cached_image_warning_days_reads_value(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            store.set_cached_image_stale_warning_days(7)
+            self.assertEqual(7, store.cached_image_stale_warning_days())
+
+    def test_config_schema_uses_explicit_settings_columns(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "bot-bottle.db")
+            store.migrate()
+            with sqlite3.connect(store.db_path) as conn:
+                conn.row_factory = sqlite3.Row
+                columns = [
+                    row["name"]
+                    for row in conn.execute("PRAGMA table_info(bot_bottle_config)")
+                ]
+            self.assertEqual([
+                "id",
+                "cached_image_stale_warning_days",
+            ], columns)
+
+    def test_store_manager_includes_config_store(self) -> None:
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            db = Path(tmp) / "bot-bottle.db"
+            manager = StoreManager(db)
+            self.assertFalse(manager.is_migrated())
+            manager.migrate()
+            self.assertTrue(manager.is_migrated())
+
+    def test_cached_image_warning_days_returns_default_when_db_missing(self) -> None:
+        # When the db file doesn't exist yet (parent exists, file doesn't),
+        # the store returns the default without touching the file.
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            store = ConfigStore(Path(tmp) / "missing.db")
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+    def test_cached_image_warning_days_returns_default_on_null_value(self) -> None:
+        # If the row exists but the value is NULL (or not castable to int),
+        # the store falls back to the default.
+        with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
+            db_path = Path(tmp) / "bot-bottle.db"
+            store = ConfigStore(db_path)
+            store.migrate()
+            # Write a NULL value directly.
+            with sqlite3.connect(db_path) as conn:
+                conn.execute(
+                    "UPDATE bot_bottle_config SET cached_image_stale_warning_days = NULL WHERE id = 1"
+                )
+            self.assertEqual(
+                DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
+                store.cached_image_stale_warning_days(),
+            )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_consolidated_launch.py

@@ -15,6 +15,7 @@ from bot_bottle.git_gate import GitGatePlan
 from bot_bottle.orchestrator.client import RegisteredBottle
 
 _MOD = "bot_bottle.backend.docker.consolidated_launch"
+_UTIL = "bot_bottle.backend.consolidated_util"
 
 
 def _egress_plan() -> EgressPlan:
@@ -49,7 +50,7 @@ class TestLaunchConsolidated(unittest.TestCase):
                 patch(f"{_MOD}._container_ip", return_value="172.18.0.2"), \
                 patch(f"{_MOD}._network_container_ips", return_value=list(on_network)), \
                 patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.provision_git_gate", provision or Mock()):
+                patch(f"{_UTIL}.provision_git_gate", provision or Mock()):
             return launch_consolidated(_egress_plan(), _git_plan(), service=service)
 
     def test_allocates_ip_registers_and_provisions(self) -> None:
@@ -84,8 +85,8 @@ class TestLaunchConsolidated(unittest.TestCase):
 class TestTeardownConsolidated(unittest.TestCase):
     def test_deregisters_and_deprovisions(self) -> None:
         client = Mock()
-        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.deprovision_git_gate") as deprov:
+        with patch(f"{_UTIL}.OrchestratorClient", return_value=client), \
+                patch(f"{_UTIL}.deprovision_git_gate") as deprov:
             teardown_consolidated("b1", orchestrator_url="http://orch:8080")
         client.teardown_bottle.assert_called_once_with("b1")
         deprov.assert_called_once()

tests/unit/test_contrib_claude_auth.py

@@ -0,0 +1,186 @@
+"""Unit: host Claude auth extraction."""
+
+from __future__ import annotations
+
+import json
+import tempfile
+import unittest
+from datetime import datetime, timezone
+from pathlib import Path
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.contrib.claude.claude_auth import (
+    claude_auth_path,
+    claude_host_access_token,
+)
+from bot_bottle.log import Die
+
+
+def _cred_json(access_token: str, **extra: object) -> str:
+    payload: dict[str, object] = {"claudeAiOauth": {"accessToken": access_token, **extra}}
+    return json.dumps(payload)
+
+
+class TestClaudeHostAccessToken(unittest.TestCase):
+    def setUp(self):
+        self.tmp = tempfile.TemporaryDirectory(prefix="bb-claude-auth.")
+        self.home = Path(self.tmp.name)
+        self.cred_dir = self.home / ".claude"
+        self.cred_dir.mkdir()
+        self.auth_path = self.cred_dir / ".credentials.json"
+
+    def tearDown(self):
+        self.tmp.cleanup()
+
+    def _write(self, payload: dict) -> None:  # type: ignore[no-untyped-def]
+        self.auth_path.write_text(json.dumps(payload))
+
+    def test_auth_path_uses_home_env(self):
+        self.assertEqual(
+            self.auth_path,
+            claude_auth_path({"HOME": str(self.home)}),
+        )
+
+    # --- file-based (Linux) ---
+
+    def test_file_returns_access_token(self):
+        key = "sk-ant-oat01-real-key"  # gitleaks:allow
+        self._write({"claudeAiOauth": {"accessToken": key}})
+        out = claude_host_access_token({"HOME": str(self.home)})
+        self.assertEqual(key, out)
+
+    def test_file_missing_claude_ai_oauth_dies(self):
+        self._write({"hasCompletedOnboarding": True})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_file_missing_access_token_dies(self):
+        self._write({"claudeAiOauth": {"expiresAt": 2000000000000}})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_file_empty_access_token_dies(self):
+        self._write({"claudeAiOauth": {"accessToken": ""}})
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_file_expired_token_dies(self):
+        # expiresAt is milliseconds; 1_000_000 ms is year 1970
+        self._write({
+            "claudeAiOauth": {"accessToken": "sk-ant-oat01-x", "expiresAt": 1_000_000},  # gitleaks:allow
+        })
+        with self.assertRaises(Die):
+            claude_host_access_token(
+                {"HOME": str(self.home)},
+                now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+            )
+
+    def test_file_future_expiry_is_accepted(self):
+        key = "sk-ant-oat01-y"  # gitleaks:allow
+        # 2_000_000_000_000 ms ≈ year 2033
+        self._write({
+            "claudeAiOauth": {"accessToken": key, "expiresAt": 2_000_000_000_000},
+        })
+        out = claude_host_access_token(
+            {"HOME": str(self.home)},
+            now=datetime(2026, 1, 1, tzinfo=timezone.utc),
+        )
+        self.assertEqual(key, out)
+
+    def test_file_absent_expiry_is_accepted(self):
+        key = "sk-ant-oat01-z"  # gitleaks:allow
+        self._write({"claudeAiOauth": {"accessToken": key}})
+        out = claude_host_access_token({"HOME": str(self.home)})
+        self.assertEqual(key, out)
+
+    def test_file_non_json_dies(self):
+        self.auth_path.write_text("not json {{{")
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_file_json_array_root_dies(self):
+        self.auth_path.write_text("[]")
+        with self.assertRaises(Die):
+            claude_host_access_token({"HOME": str(self.home)})
+
+    def test_file_extra_fields_are_ignored(self):
+        key = "sk-ant-oat01-real"  # gitleaks:allow
+        self._write({
+            "claudeAiOauth": {
+                "accessToken": key,
+                "refreshToken": "sk-ant-ort01-secret",  # gitleaks:allow
+                "scopes": ["user:inference"],
+                "expiresAt": 2_000_000_000_000,
+            },
+        })
+        out = claude_host_access_token({"HOME": str(self.home)})
+        self.assertEqual(key, out)
+
+    # --- macOS Keychain fallback ---
+
+    def _home_without_creds(self) -> Path:
+        """A home dir that has .claude/ but no .credentials.json."""
+        empty = self.home / "no-creds"
+        (empty / ".claude").mkdir(parents=True)
+        return empty
+
+    def _mock_keychain(self, stdout: str, returncode: int = 0) -> MagicMock:
+        mock = MagicMock()
+        mock.returncode = returncode
+        mock.stdout = stdout
+        return mock
+
+    def test_keychain_used_when_file_absent(self):
+        key = "sk-ant-oat01-keychain"  # gitleaks:allow
+        home = self._home_without_creds()
+        with patch(
+            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
+            return_value=self._mock_keychain(_cred_json(key)),
+        ), patch(
+            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
+        ):
+            out = claude_host_access_token({"HOME": str(home)})
+        self.assertEqual(key, out)
+
+    def test_keychain_failure_when_file_absent_dies(self):
+        home = self._home_without_creds()
+        with patch(
+            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
+            return_value=self._mock_keychain("", returncode=44),
+        ), patch(
+            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
+        ):
+            with self.assertRaises(Die):
+                claude_host_access_token({"HOME": str(home)})
+
+    def test_no_file_no_keychain_on_linux_dies(self):
+        home = self._home_without_creds()
+        with patch("bot_bottle.contrib.claude.claude_auth.sys.platform", "linux"):
+            with self.assertRaises(Die):
+                claude_host_access_token({"HOME": str(home)})
+
+    def test_keychain_non_json_dies(self):
+        home = self._home_without_creds()
+        with patch(
+            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
+            return_value=self._mock_keychain("not-json"),
+        ), patch(
+            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
+        ):
+            with self.assertRaises(Die):
+                claude_host_access_token({"HOME": str(home)})
+
+    def test_keychain_security_not_found_dies(self):
+        home = self._home_without_creds()
+        with patch(
+            "bot_bottle.contrib.claude.claude_auth.subprocess.run",
+            side_effect=FileNotFoundError,
+        ), patch(
+            "bot_bottle.contrib.claude.claude_auth.sys.platform", "darwin",
+        ):
+            with self.assertRaises(Die):
+                claude_host_access_token({"HOME": str(home)})
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_docker_launch_committed_image.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import contextlib
+import dataclasses
 import io
 import tempfile
 import unittest
@@ -18,6 +19,7 @@ from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.docker.consolidated_launch import LaunchContext
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
+from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 from tests.unit import use_bottle_root
 
@@ -92,6 +94,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
                 mock.patch.object(launch_mod.docker_mod, "image_exists", return_value=image_present), \
                 mock.patch.object(launch_mod.docker_mod, "build_image", side_effect=_build), \
                 mock.patch.object(launch_mod.docker_mod, "verify_agent_image"), \
+                mock.patch.object(launch_mod.docker_mod, "image_created_at"), \
                 mock.patch.object(launch_mod, "launch_consolidated", return_value=_CTX), \
                 mock.patch.object(launch_mod, "teardown_consolidated"), \
                 mock.patch.object(launch_mod, "DockerGateway", return_value=gw), \
@@ -112,7 +115,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
         with self._patched(
             committed_tag=committed_tag, image_present=image_present, compose=compose,
         ) as built:
-            with launch_mod.launch(plan, provision=mock.Mock(return_value=None)):
+            images = launch_mod.build_or_load_images(plan)
+            with launch_mod.launch(plan, images, provision=mock.Mock(return_value=None)):
                 pass
         return built
 
@@ -127,14 +131,34 @@ class TestLaunchCommittedImage(unittest.TestCase):
             captured.append(p)
             return {"services": {"agent": {}}}
 
-        with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose):
-            with launch_mod.launch(_plan(self._tmp), provision=mock.Mock(return_value=None)):
+        with self._patched(committed_tag=_COMMITTED_TAG, image_present=True, compose=compose) as _:
+            plan = _plan(self._tmp)
+            images = launch_mod.build_or_load_images(plan)
+            with launch_mod.launch(plan, images, provision=mock.Mock(return_value=None)):
                 pass
         self.assertEqual(_COMMITTED_TAG, captured[0].image)
 
     def test_falls_back_to_build_when_no_committed_image(self) -> None:
         self.assertEqual([_DEFAULT_IMAGE], self._run_launch(_plan(self._tmp), committed_tag=None))
 
+    def test_cached_images_skip_build_when_present(self) -> None:
+        base = _plan(self._tmp)
+        plan = dataclasses.replace(
+            base,
+            spec=dataclasses.replace(base.spec, image_policy="cached"),
+        )
+        built = self._run_launch(plan, committed_tag=None, image_present=True)
+        self.assertEqual([], built)
+
+    def test_cached_images_die_when_agent_missing(self) -> None:
+        base = _plan(self._tmp)
+        plan = dataclasses.replace(
+            base,
+            spec=dataclasses.replace(base.spec, image_policy="cached"),
+        )
+        with self.assertRaises(Die):
+            self._run_launch(plan, committed_tag=None, image_present=False)
+
     def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
         built = self._run_launch(_plan(self._tmp), committed_tag=_COMMITTED_TAG, image_present=False)
         self.assertEqual([_DEFAULT_IMAGE], built)

tests/unit/test_docker_launch_teardown.py

@@ -16,7 +16,7 @@ from pathlib import Path
 from unittest import mock
 
 from bot_bottle.agent_provider import AgentProvisionPlan
-from bot_bottle.backend import BottleSpec
+from bot_bottle.backend import BottleImages, BottleSpec
 from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.backend.docker.consolidated_launch import LaunchContext
@@ -93,6 +93,8 @@ class TestTeardownWarning(unittest.TestCase):
             orchestrator_url="http://orch:8099",
         )
 
+        images = BottleImages(agent="bot-bottle-claude:latest", sidecar="bot-bottle-sidecars:latest")
+
         with mock.patch.object(launch_mod.docker_mod, "build_image"), \
              mock.patch.object(launch_mod.docker_mod, "verify_agent_image"), \
              mock.patch.object(launch_mod, "launch_consolidated", return_value=ctx), \
@@ -113,7 +115,7 @@ class TestTeardownWarning(unittest.TestCase):
              ), \
              contextlib.redirect_stderr(buf):
             provision = mock.Mock(return_value=None)
-            with launch_mod.launch(plan, provision=provision):
+            with launch_mod.launch(plan, images, provision=provision):
                 pass
 
         output = buf.getvalue()

tests/unit/test_docker_provision_git_user.py

@@ -45,12 +45,15 @@ _PROVIDER = _Provider()
 
 
 def _plan(*, git_user: dict | None = None,  # type: ignore
+          git_repos: dict | None = None,  # type: ignore
           copy_cwd: bool = False,
           user_cwd: str = "/tmp/x",
           stage_dir: Path | None = None) -> DockerBottlePlan:
     bottle_json: dict = {}  # type: ignore
     if git_user is not None:
         bottle_json["git-gate"] = {"user": git_user}
+    if git_repos is not None:
+        bottle_json.setdefault("git-gate", {})["repos"] = git_repos
     index = ManifestIndex.from_json_obj({
         "bottles": {"dev": bottle_json},
         "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
@@ -125,6 +128,62 @@ class TestProvisionGitUser(unittest.TestCase):
         _PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
         self.assertEqual([], _git_config_exec_calls(bottle))
 
+    def test_repairs_git_xdg_directory_for_runtime_user(self):
+        bottle = _make_bottle()
+        _PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
+
+        script, user = next(
+            (call.args[0], call.kwargs.get("user", "node"))
+            for call in bottle.exec.call_args_list
+            if "/home/node/.config/git" in call.args[0]
+        )
+        self.assertEqual("root", user)
+        self.assertIn("chown node:node /home/node", script)
+        self.assertIn("chmod 755 /home/node", script)
+        self.assertIn("mkdir -p /home/node/.config/git", script)
+        self.assertIn("chown -R node:node /home/node/.config", script)
+        self.assertIn("chmod -R u+rwX,go+rX /home/node/.config", script)
+
+    def test_fails_closed_when_home_permissions_cannot_be_repaired(self):
+        bottle = _make_bottle()
+        bottle.exec.return_value = ExecResult(1, "", "read-only filesystem")
+
+        with self.assertRaises(SystemExit):
+            _PROVIDER.provision_git(bottle, _plan(stage_dir=self.stage))
+
+    def _git_plan(self) -> DockerBottlePlan:
+        return _plan(
+            git_repos={
+                "repo": {
+                    "url": "ssh://git@example.com/repo.git",
+                    "key": {"provider": "static", "path": "/dev/null"},
+                    "host_key": "ssh-ed25519 AAAA",
+                },
+            },
+            stage_dir=self.stage,
+        )
+
+    def test_fails_closed_when_gitconfig_permissions_cannot_be_set(self):
+        bottle = _make_bottle()
+        bottle.exec.side_effect = [
+            ExecResult(0, "", ""),
+            ExecResult(1, "", "chown failed"),
+        ]
+
+        with self.assertRaises(SystemExit):
+            _PROVIDER.provision_git(bottle, self._git_plan())
+
+    def test_fails_closed_when_runtime_user_cannot_read_gitconfig(self):
+        bottle = _make_bottle()
+        bottle.exec.side_effect = [
+            ExecResult(0, "", ""),
+            ExecResult(0, "", ""),
+            ExecResult(1, "", "permission denied"),
+        ]
+
+        with self.assertRaises(SystemExit):
+            _PROVIDER.provision_git(bottle, self._git_plan())
+
     def test_sets_name_and_email(self):
         plan = _plan(
             git_user={"name": "Eric Bauerfeld", "email": "eric@dideric.is"},

tests/unit/test_docker_util_image.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 
 import subprocess
 import unittest
+from datetime import timezone
 from unittest.mock import patch
 
 from bot_bottle.backend.docker import util as docker_mod
@@ -26,6 +27,59 @@ def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
     )
 
 
+
+class TestImageCreatedAt(unittest.TestCase):
+    def test_parses_docker_timestamp_with_nanoseconds(self):
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout="2026-07-06T15:33:47.123456789Z\n"),
+        ) as run:
+            created = docker_mod.image_created_at("bot-bottle-claude:latest")
+        self.assertIsNotNone(created)
+        assert created is not None
+        self.assertEqual(2026, created.year)
+        self.assertEqual(123456, created.microsecond)
+        self.assertEqual(timezone.utc, created.tzinfo)
+        self.assertEqual(
+            ["docker", "image", "inspect", "--format", "{{.Created}}", "bot-bottle-claude:latest"],
+            run.call_args.args[0],
+        )
+
+    def test_dies_on_inspect_failure(self):
+        with patch.object(
+            docker_mod.subprocess, "run", return_value=_fail("No such image"),
+        ), patch.object(
+            docker_mod, "die", side_effect=SystemExit("die"),
+        ) as die:
+            with self.assertRaises(SystemExit):
+                docker_mod.image_created_at("missing:tag")
+        die.assert_called_once()
+        self.assertIn("missing:tag", die.call_args.args[0])
+
+    def test_returns_none_on_invalid_timestamp(self):
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout="not-a-timestamp\n"),
+        ):
+            result = docker_mod.image_created_at("some:tag")
+        self.assertIsNone(result)
+
+    def test_returns_none_on_empty_stdout(self):
+        with patch.object(
+            docker_mod.subprocess, "run",
+            return_value=_ok(stdout=""),
+        ):
+            result = docker_mod.image_created_at("some:tag")
+        self.assertIsNone(result)
+
+    def test_parse_docker_timestamp_no_tzinfo_defaults_to_utc(self):
+        # A bare datetime with no tz offset should be treated as UTC.
+        dt = docker_mod._parse_docker_timestamp("2024-05-01T10:00:00.000000")
+        self.assertIsNotNone(dt.tzinfo)
+        self.assertEqual(timezone.utc, dt.tzinfo)
+
+
+
 class TestCommitContainer(unittest.TestCase):
     def test_runs_docker_commit(self):
         with patch.object(

tests/unit/test_egress_addon_request_flow.py

@@ -413,6 +413,28 @@ class TestAuthInjection(unittest.TestCase):
         assert flow.response is not None
         self.assertEqual(403, flow.response.status_code)
 
+    def test_preserve_auth_passes_agent_token_through(self) -> None:
+        route = Route(host="registry-1.docker.io", preserve_auth=True)
+        addon = _addon(Config(routes=(route,)))
+        flow = _Flow(_Request(
+            host="registry-1.docker.io",
+            headers={"authorization": "Bearer agent-registry-token"},
+        ))
+        _run_request(addon, flow)
+        self.assertEqual("Bearer agent-registry-token", flow.request.headers.get("authorization"))
+        self.assertIsNone(flow.response)
+
+    def test_default_route_strips_agent_auth(self) -> None:
+        route = Route(host="registry-1.docker.io")
+        addon = _addon(Config(routes=(route,)))
+        flow = _Flow(_Request(
+            host="registry-1.docker.io",
+            headers={"authorization": "Bearer agent-registry-token"},
+        ))
+        _run_request(addon, flow)
+        self.assertIsNone(flow.request.headers.get("authorization"))
+        self.assertIsNone(flow.response)
+
 
 # ---------------------------------------------------------------------------
 # git push / fetch over HTTPS

tests/unit/test_firecracker_backend.py

@@ -61,6 +61,12 @@ class TestNetpoolSlots(unittest.TestCase):
 
 
 class TestNetpoolRenderers(unittest.TestCase):
+    def test_guest_init_restores_node_home_boundary(self):
+        from bot_bottle.backend.firecracker import util
+
+        self.assertIn("chown node:node /home/node", util._GUEST_INIT)
+        self.assertIn("chmod 755 /home/node", util._GUEST_INIT)
+
     def test_nixos_module_is_non_invasive(self):
         # The NixOS module must NOT flip the host firewall backend or
         # hand interfaces to systemd-networkd; it brings the pool up via
@@ -422,10 +428,15 @@ class TestBottlePlanProperties(unittest.TestCase):
         ap.command = "claude"
         ap.prompt_mode = "append_file"
         ap.template = "claude"
+        ap.guest_home = "/home/node"
+        ap.guest_env = {}
+        egress_plan = cast(Any, MagicMock())
+        egress_plan.canary = ""
+        egress_plan.canary_env = ""
         fields = dict(
             spec=cast(Any, MagicMock()), manifest=cast(Any, MagicMock()),
             stage_dir=Path("/stage"), git_gate_plan=cast(Any, MagicMock()),
-            egress_plan=cast(Any, MagicMock()), supervise_plan=None,
+            egress_plan=egress_plan, supervise_plan=None,
             agent_provision=ap, slug="demo-x", forwarded_env={},
         )
         fields.update(overrides)
@@ -451,6 +462,12 @@ class TestBottlePlanProperties(unittest.TestCase):
         self.assertEqual("10.243.0.0:9420", p.git_gate_insteadof_host)
         self.assertEqual("http", p.git_gate_insteadof_scheme)
 
+    def test_guest_env_pins_global_git_config(self):
+        from bot_bottle.backend.firecracker.launch import _agent_guest_env
+
+        env = _agent_guest_env(self._plan(), "10.243.0.0")
+        self.assertEqual("/home/node/.gitconfig", env["GIT_CONFIG_GLOBAL"])
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_firecracker_image_builder.py

@@ -35,6 +35,15 @@ class TestBuildAgentRootfsDir(unittest.TestCase):
         build.assert_not_called()
         self.assertEqual(base, out)
 
+    def test_cached_lookup_requires_ready_marker(self):
+        digest = image_builder._rootfs_digest(self.dockerfile)
+        base = self.cache / "rootfs" / f"agent-{digest}"
+        base.mkdir(parents=True)
+        with patch.object(image_builder.util, "cache_dir", return_value=self.cache):
+            self.assertIsNone(image_builder.cached_agent_rootfs_dir(self.dockerfile))
+            (base / ".bb-ready").write_text("ok\n")
+            self.assertEqual(base, image_builder.cached_agent_rootfs_dir(self.dockerfile))
+
     def test_cache_miss_builds_injects_and_marks_ready(self):
         with patch.object(image_builder.util, "cache_dir", return_value=self.cache), \
              patch.object(image_builder, "_build_in_infra") as build, \

tests/unit/test_firecracker_infra_vm.py

@@ -71,6 +71,16 @@ class TestSshGatewayTransport(unittest.TestCase):
             t.exec(["mkdir", "-p", "/git-gate"])
 
 
+class TestGatewayCaPem(unittest.TestCase):
+    def test_dies_when_cert_never_appears(self) -> None:
+        from subprocess import CompletedProcess
+        infra = infra_vm.InfraVm(vm=None, guest_ip="10.0.0.1", private_key=Path("/k"))
+        with patch.object(infra_vm.subprocess, "run",
+                          return_value=CompletedProcess([], 1, stdout="", stderr="")), \
+             self.assertRaises(SystemExit):
+            infra.gateway_ca_pem(timeout=0)
+
+
 class TestRegistryVolume(unittest.TestCase):
     def test_reuses_existing_volume(self):
         import tempfile

tests/unit/test_image_cache.py

@@ -0,0 +1,81 @@
+"""Unit: image_cache.py — check_stale / check_stale_path."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from datetime import datetime, timedelta, timezone
+from pathlib import Path
+from unittest.mock import patch
+
+from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path
+
+
+class TestCheckStale(unittest.TestCase):
+    def _run(self, threshold: int, age_days: float) -> None:
+        created = datetime.now(tz=timezone.utc) - timedelta(days=age_days)
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = threshold
+            check_stale("test image", created)
+
+    def test_negative_threshold_never_raises(self):
+        # Threshold < 0 means the check is disabled — always passes.
+        self._run(threshold=-1, age_days=9999)
+
+    def test_zero_threshold_raises_immediately(self):
+        # threshold=0 means any image is stale the moment it exists.
+        with self.assertRaises(StaleImageError):
+            self._run(threshold=0, age_days=0.1)
+
+    def test_within_threshold_does_not_raise(self):
+        # Age well under threshold — should pass silently.
+        self._run(threshold=7, age_days=2)
+
+    def test_at_threshold_does_not_raise(self):
+        # Exactly at the boundary is fine (<=, not <).
+        self._run(threshold=1, age_days=0.9999)
+
+    def test_exceeds_threshold_raises(self):
+        created = datetime.now(tz=timezone.utc) - timedelta(days=3)
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = 1
+            with self.assertRaises(StaleImageError) as ctx:
+                check_stale("agent image 'bot-bottle:latest'", created)
+        self.assertIn("agent image", str(ctx.exception))
+        self.assertIn("day(s) old", str(ctx.exception))
+
+    def test_naive_datetime_treated_as_utc(self):
+        # check_stale calls .astimezone(utc) on the input; naive datetimes
+        # that would be interpreted as local time should still work.
+        # We can't control the local tz in a unit test, so just ensure
+        # no exception is thrown for a very recent naive datetime.
+        naive_now = datetime(2099, 1, 1)  # far future, always "fresh"
+        with patch("bot_bottle.image_cache.ConfigStore") as cs:
+            cs.return_value.cached_image_stale_warning_days.return_value = 1
+            # Should not raise — the image is brand new.
+            check_stale("test image", naive_now)
+
+
+class TestCheckStalePath(unittest.TestCase):
+    def test_delegates_to_check_stale_with_mtime(self):
+        with tempfile.NamedTemporaryFile() as f:
+            path = Path(f.name)
+            with patch("bot_bottle.image_cache.check_stale") as mock_check:
+                check_stale_path("some artifact", path)
+            mock_check.assert_called_once()
+            label, dt = mock_check.call_args.args
+            self.assertEqual("some artifact", label)
+            self.assertIsInstance(dt, datetime)
+            self.assertIsNotNone(dt.tzinfo)
+
+    def test_raises_stale_for_old_file(self):
+        with tempfile.NamedTemporaryFile() as f:
+            path = Path(f.name)
+            with patch("bot_bottle.image_cache.ConfigStore") as cs:
+                cs.return_value.cached_image_stale_warning_days.return_value = 0
+                with self.assertRaises(StaleImageError):
+                    check_stale_path("cached artifact", path)
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_infra_artifact.py

@@ -96,7 +96,7 @@ class TestVersionInputs(unittest.TestCase):
         (pkg / "app.py").write_text("print('hi')\n")
         (pkg / "egress_entrypoint.sh").write_text("#!/bin/sh\nexec mitmdump\n")
         (pkg / "netpool.defaults.env").write_text("FOO=1\n")
-        for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra"):
+        for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra", "Dockerfile.infra.fc"):
             (root / name).write_text(f"FROM scratch  # {name}\n")
         (root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
 

tests/unit/test_macos_consolidated_launch.py

@@ -17,6 +17,7 @@ from bot_bottle.git_gate import GitGatePlan
 from bot_bottle.orchestrator.client import RegisteredBottle
 
 _MOD = "bot_bottle.backend.macos_container.consolidated_launch"
+_UTIL = "bot_bottle.backend.consolidated_util"
 
 
 def _egress_plan() -> EgressPlan:
@@ -87,7 +88,7 @@ class TestRegisterAgent(unittest.TestCase):
         *, source_ip: str = "192.168.128.9",
     ):
         with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.provision_git_gate", provision or Mock()), \
+                patch(f"{_UTIL}.provision_git_gate", provision or Mock()), \
                 patch(f"{_MOD}.live_source_ips", return_value=[]):
             return register_agent(
                 _egress_plan(), _git_plan(),
@@ -126,8 +127,8 @@ class TestTeardown(unittest.TestCase):
     def test_deregisters_and_deprovisions(self) -> None:
         client = Mock()
         deprovision = Mock()
-        with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.deprovision_git_gate", deprovision):
+        with patch(f"{_UTIL}.OrchestratorClient", return_value=client), \
+                patch(f"{_UTIL}.deprovision_git_gate", deprovision):
             teardown_consolidated("b1", orchestrator_url="http://o:8099")
         client.teardown_bottle.assert_called_once_with("b1")
         self.assertEqual("b1", deprovision.call_args.args[1])
@@ -190,7 +191,7 @@ class TestRegisterAgentReconciles(unittest.TestCase):
 
     def _register(self, client: Mock) -> None:
         with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.provision_git_gate"), \
+                patch(f"{_UTIL}.provision_git_gate"), \
                 patch(f"{_MOD}.live_source_ips", return_value=["10.0.0.7"]):
             register_agent(
                 _egress_plan(), _git_plan(),
@@ -228,7 +229,7 @@ class TestRegisterAgentReconciles(unittest.TestCase):
         from bot_bottle.backend.macos_container.enumerate import EnumerationError
         client = _client()
         with patch(f"{_MOD}.OrchestratorClient", return_value=client), \
-                patch(f"{_MOD}.provision_git_gate"), \
+                patch(f"{_UTIL}.provision_git_gate"), \
                 patch(f"{_MOD}.live_source_ips",
                       side_effect=EnumerationError("container list failed")):
             register_agent(

tests/unit/test_macos_container_launch_wiring.py

@@ -12,7 +12,7 @@ import unittest
 from pathlib import Path
 from types import SimpleNamespace
 from typing import cast
-from unittest.mock import patch
+from unittest.mock import ANY, patch
 
 from bot_bottle.backend.macos_container.bottle import MacosContainerBottle
 from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
@@ -21,8 +21,9 @@ from bot_bottle.backend.macos_container.gateway_hosts import GATEWAY_HOSTNAME
 from bot_bottle.backend.macos_container.launch import (
     _agent_run_argv,
     _identity_proxy_env,
+    build_or_load_images,
 )
-from bot_bottle.backend.macos_container.rootless_podman import guest_env
+from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 
 _BOTTLE = "bot_bottle.backend.macos_container.bottle"
@@ -47,6 +48,7 @@ def _plan(
     *,
     agent_git_gate_url: str = "",
     agent_supervise_url: str = "",
+    image_policy: str = "fresh",
 ) -> MacosContainerBottlePlan:
     routes_path = stage_dir / "routes.yaml"
     routes_path.write_text("routes: []\n", encoding="utf-8")
@@ -61,12 +63,13 @@ def _plan(
         canary_env="",
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(),
+        spec=SimpleNamespace(image_policy=image_policy),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
         container_name="bot-bottle-dev-abc",
         image="bot-bottle-agent:latest",
+        dockerfile_path="/repo/Dockerfile",
         forwarded_env={"OAUTH_TOKEN": "host-value"},
         egress_plan=egress_plan,
         git_gate_plan=SimpleNamespace(upstreams=()),
@@ -77,10 +80,91 @@ def _plan(
         ),
         agent_git_gate_url=agent_git_gate_url,
         agent_supervise_url=agent_supervise_url,
-        docker_access=False,
     ))
 
 
+class TestBuildOrLoadImages(unittest.TestCase):
+    def setUp(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory()
+        self.plan = _plan(Path(self._tmp.name))
+
+    def tearDown(self) -> None:
+        self._tmp.cleanup()
+
+    def test_reuses_present_committed_image(self) -> None:
+        with (
+            patch(
+                "bot_bottle.backend.macos_container.launch.read_committed_image",
+                return_value="committed:latest",
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.image_exists",
+                return_value=True,
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.build_image"
+            ) as build,
+        ):
+            images = build_or_load_images(self.plan)
+
+        self.assertEqual("committed:latest", images.agent)
+        build.assert_not_called()
+
+    def test_reuses_present_cached_image(self) -> None:
+        plan = _plan(Path(self._tmp.name), image_policy="cached")
+        with (
+            patch(
+                "bot_bottle.backend.macos_container.launch.read_committed_image",
+                return_value=None,
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.image_exists",
+                return_value=True,
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.build_image"
+            ) as build,
+        ):
+            images = build_or_load_images(plan)
+
+        self.assertEqual(plan.image, images.agent)
+        build.assert_not_called()
+
+    def test_cached_policy_rejects_missing_image(self) -> None:
+        plan = _plan(Path(self._tmp.name), image_policy="cached")
+        with (
+            patch(
+                "bot_bottle.backend.macos_container.launch.read_committed_image",
+                return_value=None,
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.image_exists",
+                return_value=False,
+            ),
+        ):
+            with self.assertRaises(Die):
+                build_or_load_images(plan)
+
+    def test_fresh_policy_builds_image(self) -> None:
+        with (
+            patch(
+                "bot_bottle.backend.macos_container.launch.read_committed_image",
+                return_value=None,
+            ),
+            patch(
+                "bot_bottle.backend.macos_container.launch.container_mod.build_image"
+            ) as build,
+        ):
+            images = build_or_load_images(self.plan)
+
+        self.assertEqual(self.plan.image, images.agent)
+        build.assert_called_once_with(
+            self.plan.image,
+            ANY,
+            dockerfile=self.plan.dockerfile_path,
+        )
+
+
 class TestAgentRunArgv(unittest.TestCase):
     def setUp(self) -> None:
         self._tmp = tempfile.TemporaryDirectory()
@@ -180,18 +264,6 @@ class TestIdentityTokenDelivery(unittest.TestCase):
         self.assertNotIn("--env", argv)
 
 
-class TestRootlessPodmanEnvironment(unittest.TestCase):
-    def test_disabled_bottle_gets_no_docker_environment(self) -> None:
-        self.assertEqual({}, guest_env(False))
-
-    def test_enabled_bottle_uses_only_guest_local_socket(self) -> None:
-        env = guest_env(True)
-        self.assertEqual(
-            "unix:///tmp/bot-bottle-podman-run/podman.sock", env["DOCKER_HOST"],
-        )
-        self.assertNotIn("/var/run/docker.sock", " ".join(env.values()))
-
-
 class TestPlanIdentityToken(unittest.TestCase):
     """git-gate's gitconfig extraHeader and the supervise MCP --header read
     `getattr(plan, "identity_token", "")` at provision time and both bypass the

tests/unit/test_macos_container_util.py

@@ -391,5 +391,72 @@ class TestWaitContainerIpv4(unittest.TestCase):
             self.assertEqual("", util.wait_container_ipv4_on_network("c", "net", timeout=-1))
 
 
+class TestMacosContainerImageCreatedAt(unittest.TestCase):
+    def _ok(self, stdout: str) -> "util.subprocess.CompletedProcess":  # type: ignore
+        return util.subprocess.CompletedProcess(
+            args=[], returncode=0, stdout=stdout, stderr="",
+        )
+
+    def _fail(self, stderr: str = "no such image") -> "util.subprocess.CompletedProcess":  # type: ignore
+        return util.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr=stderr,
+        )
+
+    def test_parses_iso_timestamp_from_dict(self):
+        payload = '[{"created": "2025-06-01T12:00:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("bot-bottle-agent:latest")
+        self.assertIsNotNone(dt)
+        assert dt is not None
+        self.assertEqual(2025, dt.year)
+        self.assertEqual(6, dt.month)
+        self.assertEqual(1, dt.day)
+
+    def test_accepts_list_or_dict_input(self):
+        # Container CLI may return a list; we take the first element.
+        payload = '[{"created": "2024-01-15T08:30:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("some-image:latest")
+        self.assertIsNotNone(dt)
+        assert dt is not None
+        self.assertEqual(2024, dt.year)
+
+    def test_accepts_uppercase_Created_field(self):
+        payload = '[{"Created": "2024-03-20T10:00:00"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            dt = util.image_created_at("some-image:latest")
+        self.assertIsNotNone(dt)
+        assert dt is not None
+        self.assertEqual(2024, dt.year)
+        self.assertEqual(3, dt.month)
+
+    def test_dies_on_nonzero_returncode(self):
+        with patch.object(util.subprocess, "run", return_value=self._fail("not found")), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("missing:tag")
+        die.assert_called_once()
+        self.assertIn("missing:tag", die.call_args.args[0])
+
+    def test_dies_on_malformed_json(self):
+        with patch.object(util.subprocess, "run", return_value=self._ok("not-json {")), \
+             patch.object(util, "die", side_effect=SystemExit("die")) as die:
+            with self.assertRaises(SystemExit):
+                util.image_created_at("some:tag")
+        die.assert_called_once()
+
+    def test_returns_none_when_no_created_field(self):
+        payload = '[{"id": "sha256:abc123"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            result = util.image_created_at("some:tag")
+        self.assertIsNone(result)
+
+    def test_returns_none_on_invalid_timestamp_format(self):
+        payload = '[{"created": "not-a-date"}]'
+        with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
+            result = util.image_created_at("some:tag")
+        self.assertIsNone(result)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_infra.py

@@ -61,6 +61,20 @@ class TestInfraRun(unittest.TestCase):
         mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
         self.assertTrue(all("bot-bottle.db" not in m for m in mounts))
 
+    def test_ca_is_persisted_on_the_host_not_the_container_volume(self) -> None:
+        """The CA survives infra recreation and cannot be removed by Apple
+        Container's volume-prune command."""
+        argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
+        mounts = [argv[i + 1] for i, a in enumerate(argv) if a == "--mount"]
+        ca_mounts = [
+            m for m in mounts
+            if "target=/home/mitmproxy/.mitmproxy" in m
+        ]
+        self.assertEqual(1, len(ca_mounts))
+        self.assertIn("source=", ca_mounts[0])
+        self.assertIn("/gateway-ca", ca_mounts[0])
+        self.assertNotIn(",readonly", ca_mounts[0])
+
     def test_nat_network_precedes_the_host_only_network(self) -> None:
         argv = self._run_container(MacosInfraService(repo_root=Path("/r")))
         nets = [argv[i + 1] for i, a in enumerate(argv) if a == "--network"]
@@ -153,6 +167,14 @@ class TestCaCertPem(unittest.TestCase):
         argv = mod.run_container_argv.call_args.args[0]
         self.assertEqual(["container", "exec", "bot-bottle-mac-infra", "cat"], argv[:4])
 
+    def test_raises_gateway_error_when_cert_never_appears(self) -> None:
+        from bot_bottle.backend.macos_container.gateway import GatewayError
+        svc = MacosInfraService(repo_root=Path("/r"))
+        with patch(f"{_INFRA}.container_mod") as mod:
+            mod.run_container_argv.return_value = _fail()
+            with self.assertRaises(GatewayError):
+                svc.ca_cert_pem(timeout=0)
+
 
 class TestProbeControlPlane(unittest.TestCase):
     def test_returns_url_when_running(self) -> None: