test: replace /bin/sleep with portable Python sleep in test_gateway_init

/bin/sleep does not exist at FHS paths on NixOS (the KVM self-hosted
runner's OS). All 13 FileNotFoundError failures in TestSupervisor were
caused by _DaemonSpec tuples spawning /bin/sleep directly. Replace with
a module-level _SLEEP_30/_SLEEP_60 tuple using sys.executable so the
supervisor tests run on any platform. Also fix TestMainEndToEnd._run()
and remove the now-unnecessary setUpClass guard.

.gitea/workflows/test.yml

@@ -25,68 +25,15 @@ on:
       - '.gitea/workflows/**.yml'
       - 'scripts/**'
       - 'README.md'
-      # Dockerfiles and pyproject.toml are baked into the infra rootfs; a
-      # change here alters what the integration/coverage jobs build locally.
-      - 'Dockerfile*'
-      - 'pyproject.toml'
   pull_request:
     paths:
       - '**.py'
       - '.gitea/workflows/**.yml'
       - 'scripts/**'
       - 'README.md'
-      - 'Dockerfile*'
-      - 'pyproject.toml'
   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:
@@ -136,10 +83,9 @@ jobs:
   # PRs don't execute untrusted code on the privileged runner.
   #
   # Runner prerequisites (provision once; see README "Firecracker on Linux"):
-  # `firecracker` on PATH, `/dev/kvm` accessible, cached kernel +
+  # `firecracker` on PATH, `/dev/kvm` accessible, Docker, cached kernel +
   # static dropbear, and the pool as a persistent systemd unit.
   integration-firecracker:
-    needs: build-infra
     runs-on: [self-hosted, kvm]
     if: >-
       github.event_name == 'push' ||
@@ -159,15 +105,6 @@ 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: 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
@@ -175,7 +112,6 @@ jobs:
       - name: Run integration tests (firecracker)
         env:
           BOT_BOTTLE_BACKEND: firecracker
-          BOT_BOTTLE_INFRA_ARTIFACT_DIR: ${{ github.workspace }}/infra-candidate
         run: python3 -m unittest discover -t . -s tests/integration -v
 
   # Combined unit+integration coverage + the diff-coverage gate (the hard
@@ -194,12 +130,7 @@ jobs:
   #
   # 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.
   coverage:
-    needs: [build-infra, integration-firecracker]
     timeout-minutes: 15
     runs-on: [self-hosted, kvm]
     if: >-
@@ -222,52 +153,14 @@ jobs:
           # range overlap; it prints the exact `backend setup` fix.
           python3 cli.py backend status --backend=firecracker
 
-      - name: Download the candidate already exercised by integration
-        uses: actions/download-artifact@v3
-        with:
-          name: infra-candidate
-          path: infra-candidate
-
       # 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
 
       - name: Diff-coverage gate (changed lines >= 90%)
         run: |
           git fetch --no-tags origin main:refs/remotes/origin/main
           python3 scripts/diff_coverage.py --base origin/main --min 90
-
-  publish-infra:
-    needs: [stage-firecracker-inputs, build-infra, 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
-        uses: actions/download-artifact@v3
-        with:
-          name: infra-candidate
-          path: infra-candidate
-
-      # 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)
-        uses: actions/download-artifact@v3
-        with:
-          name: firecracker-inputs
-          path: firecracker-inputs
-
-      - name: Publish the tested candidate
-        env:
-          BOT_BOTTLE_INFRA_ARTIFACT_TOKEN: ${{ secrets.BOT_BOTTLE_INFRA_ARTIFACT_TOKEN }}
-          BOT_BOTTLE_FC_DROPBEAR: ${{ github.workspace }}/firecracker-inputs/dropbear
-        run: python3 -m bot_bottle.backend.firecracker.publish_infra --publish-dir infra-candidate

Dockerfile.gateway

@@ -98,9 +98,6 @@ RUN pip install --no-cache-dir /src/
 
 # mitmdump -s requires a file path, not a module. Write a one-line shim that
 # re-exports `addons` from the installed package; mitmdump finds it there.
-# WORKDIR here also creates /app so the shim + COPYs below can write into it
-# (nothing created /app before this point).
-WORKDIR /app
 RUN printf 'from bot_bottle.egress_addon import addons\n' > /app/egress_addon.py
 COPY bot_bottle/egress_entrypoint.sh /app/egress-entrypoint.sh
 RUN chmod +x /app/egress-entrypoint.sh
@@ -120,6 +117,8 @@ RUN mkdir -p \
 # subset the bottle uses.
 EXPOSE 8888 9099 9418 9420 9100
 
+WORKDIR /app
+
 # PID 1 is the supervisor. It owns signal handling and exit-code
 # propagation; no `exec` chain in the entrypoint itself.
 ENTRYPOINT ["python3", "-m", "bot_bottle.gateway_init"]

README.md

@@ -90,7 +90,7 @@ BOT_BOTTLE_BACKEND=firecracker ./cli.py start <agent>
 
 > **NixOS:** enable `virtualisation.docker`, ensure the KVM module is loaded (`boot.kernelModules = [ "kvm-intel" ];` or `kvm-amd`), and add your user to the `kvm` and `docker` groups. For the network pool, consume the flake module — `imports = [ inputs.bot-bottle.nixosModules.firecracker-netpool ]; services.bot-bottle-firecracker = { enable = true; owner = "you"; };` — then `nixos-rebuild switch` (imperative nft/TAP rules don't survive a rebuild; channel users can `imports = [ <bot-bottle>/nix/firecracker-netpool.nix ]`). `firecracker` isn't in nixpkgs by default as a user binary — install the release binary (pin the version) and put it on `PATH`.
 
-> **CI:** the coverage gate (`.gitea/workflows/test.yml` → `coverage` job) runs on a self-hosted runner labelled `kvm`, because the Firecracker backend's VM/SSH orchestration is exercised only by the integration suite, which needs `/dev/kvm` + the provisioned pool (a container runner would skip it and read as uncovered). Provision that runner exactly like a normal Firecracker host — `firecracker` on `PATH`, `/dev/kvm`, the cached guest kernel + static dropbear, and the pool installed as the persistent systemd unit — then register it with the `kvm` label. A Docker-capable hosted job builds the candidate once; KVM tests boot those exact bytes, and a successful main run publishes them. The unit/lint jobs still run on `ubuntu-latest`.
+> **CI:** the coverage gate (`.gitea/workflows/test.yml` → `coverage` job) runs on a self-hosted runner labelled `kvm`, because the Firecracker backend's VM/SSH orchestration is exercised only by the integration suite, which needs `/dev/kvm` + the provisioned pool (a container runner would skip it and read as uncovered). Provision that runner exactly like a normal Firecracker host — `firecracker` on `PATH`, `/dev/kvm`, Docker, the cached guest kernel + static dropbear, and the pool installed as the persistent systemd unit — then register it with the `kvm` label. The unit/lint jobs still run on `ubuntu-latest`.
 
 ```sh
 ./cli.py start <agent>   # builds the image on first run, drops you into claude

bot_bottle/agent_provider.py

@@ -45,10 +45,6 @@ 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",

bot_bottle/backend/firecracker/image_builder.py

@@ -38,39 +38,25 @@ _BUILD_TIMEOUT_SECONDS = 900.0
 
 
 def _dockerfile_hash(dockerfile: Path) -> str:
-    """The Dockerfile's content hash. The shipped agent Dockerfiles COPY
-    nothing from the build context (see .dockerignore), so their content fully
-    determines the built image; a Dockerfile that adds COPY will want the
+    """Cache key: the Dockerfile's content. The shipped agent Dockerfiles
+    COPY nothing from the build context (see .dockerignore), so their content
+    fully determines the image; a Dockerfile that adds COPY will want the
     context folded in here too."""
     return hashlib.sha256(dockerfile.read_bytes()).hexdigest()[:16]
 
 
-def _rootfs_digest(dockerfile: Path) -> str:
-    """Cache key for the built AND boot-injected agent rootfs. Two inputs
-    determine the on-disk rootfs: the Dockerfile (the image) and the guest init
-    injected into it (`util._GUEST_INIT`). Folding the init in means a fix to
-    it — e.g. making /tmp world-writable — busts the cache instead of silently
-    reusing a stale rootfs built with the old init."""
-    h = hashlib.sha256()
-    h.update(_dockerfile_hash(dockerfile).encode())
-    h.update(b"\0")
-    h.update(util._GUEST_INIT.encode())
-    return h.hexdigest()[:16]
-
-
 def build_agent_rootfs_dir(
     dockerfile: Path, *, image_tag: str, smoke_test: tuple[str, ...] = (),
 ) -> Path:
     """Build `dockerfile` in the infra VM (buildah, no host docker), export its
     rootfs, inject the guest boot bits, and return the cached base dir — the
-    same shape `util.build_rootfs_ext4` consumes. Cached by Dockerfile content
-    + injected guest init, so a repeat launch skips the rebuild but an init or
-    Dockerfile change rebuilds.
+    same shape `util.build_rootfs_ext4` consumes. Cached by Dockerfile content,
+    so a repeat launch skips the rebuild.
 
     `smoke_test` (the provider's declared argv, e.g. `("claude","--version")`)
     is run in the freshly built image before export, catching an npm
     silent-failure image at build time rather than at first agent use."""
-    digest = _rootfs_digest(dockerfile)
+    digest = _dockerfile_hash(dockerfile)
     base = util.cache_dir() / "rootfs" / f"agent-{digest}"
     if (base / ".bb-ready").is_file():
         info(f"using cached agent rootfs {base.name}")

bot_bottle/backend/firecracker/infra_artifact.py

@@ -85,11 +85,6 @@ def infra_artifact_version(init_script: str, *, repo_root: Path = _REPO_ROOT) ->
         h.update(name.encode())
         h.update(b"\0")
         h.update((repo_root / name).read_bytes())
-    h.update(b"pyproject.toml\0")
-    h.update((repo_root / "pyproject.toml").read_bytes())
-    h.update(b"dropbear\0")
-    dropbear = util.dropbear_path()
-    h.update(dropbear.read_bytes() if dropbear.is_file() else b"<missing>")
     h.update(b"init\0")
     h.update(init_script.encode())
     return h.hexdigest()[:16]
@@ -116,7 +111,6 @@ def artifact_url(version: str, filename: str) -> str:
 
 _GZ_NAME = "rootfs.ext4.gz"
 _SHA_NAME = "rootfs.ext4.gz.sha256"
-_CANDIDATE_DIR_ENV = "BOT_BOTTLE_INFRA_ARTIFACT_DIR"
 
 
 def _cache_root(version: str) -> Path:
@@ -166,33 +160,6 @@ def ensure_artifact_gz(version: str) -> Path:
     """The verified, cached `rootfs.ext4.gz` for `version` — downloading it (and
     its `.sha256`) once, then reusing it. Fail-closed on a checksum mismatch:
     the partial is removed and we die rather than boot an unverified rootfs."""
-    candidate_dir = os.environ.get(_CANDIDATE_DIR_ENV, "").strip()
-    if candidate_dir:
-        root = Path(candidate_dir)
-        version_file = root / "version.txt"
-        # Guard the read so a missing version.txt is a clean error, not a raw
-        # FileNotFoundError.
-        if not version_file.is_file():
-            die(f"infra candidate bundle is incomplete: {root}")
-        declared = version_file.read_text(encoding="utf-8").strip()
-        if declared != version:
-            die(
-                f"infra candidate version mismatch: expected {version}, "
-                f"bundle contains {declared or '<empty>'}"
-            )
-        gz = root / _GZ_NAME
-        sha = root / _SHA_NAME
-        if not gz.is_file() or not sha.is_file():
-            die(f"infra candidate bundle is incomplete: {root}")
-        expected = sha.read_text().split()[0].strip().lower()
-        actual = _sha256_file(gz)
-        if actual != expected:
-            die(
-                f"infra candidate checksum mismatch for {version}:\n"
-                f"    expected {expected}\n    actual   {actual}"
-            )
-        return gz
-
     root = _cache_root(version)
     root.mkdir(parents=True, exist_ok=True)
     gz = root / _GZ_NAME

bot_bottle/backend/firecracker/infra_vm.py

@@ -161,23 +161,20 @@ def ensure_running() -> InfraVm:
     slot = netpool.orch_slot()
     url = f"http://{slot.guest_ip}:{CONTROL_PLANE_PORT}"
     key = _infra_dir() / "id_ed25519"
-    want = _expected_version()
-    if _adoptable(key, url, want):
+    if key.exists() and _health_ok(url):
         info(f"adopting running infra VM at {url}")
         return InfraVm(guest_ip=slot.guest_ip, private_key=key)
 
     with _singleton_lock():
         # Re-check under the lock: another launcher may have booted it while
         # we waited for the lock (double-checked, so we adopt not re-boot).
-        if _adoptable(key, url, want):
+        if key.exists() and _health_ok(url):
             info(f"adopting running infra VM at {url}")
             return InfraVm(guest_ip=slot.guest_ip, private_key=key)
-        # Clear a stale/hung/OUTDATED VM holding the link before booting fresh.
-        stop()
+        stop()  # clear a stale/hung VM holding the link before booting fresh
         ensure_built()
         infra = boot()
         wait_for_health(infra)
-        _record_booted_version(want)
         return infra
 
 
@@ -196,15 +193,9 @@ def _singleton_lock() -> Generator[None, None, None]:
 
 
 def stop() -> None:
-    """Stop the infra VM singleton (idempotent — absent is success). Reaps the
-    recorded VMM AND any orphaned firecracker still bound to the infra config —
-    the PID file drifts after crashes / out-of-band kills, and a survivor would
-    hold the orchestrator TAP so the next boot dies with "tap … Resource busy".
-    Drops the version marker so a stopped VM is never treated as adoptable."""
+    """Stop the infra VM singleton (idempotent — absent is success)."""
     _kill_pidfile()
-    _kill_infra_firecrackers()
     _pid_file().unlink(missing_ok=True)
-    _version_file().unlink(missing_ok=True)
 
 
 def boot() -> InfraVm:
@@ -247,36 +238,6 @@ def _pid_file() -> Path:
     return _infra_dir() / "vm.pid"
 
 
-def _version_file() -> Path:
-    """Records the infra-artifact version the *running* VM booted from, so a
-    later launcher can tell whether the singleton it found is the current code.
-    Without it, a healthy VM built from an older image gets adopted forever and
-    the new code never boots — every infra change would need an out-of-band
-    kill to dislodge the stale VM (and races whatever launched next)."""
-    return _infra_dir() / "booted-version"
-
-
-def _expected_version() -> str:
-    return infra_artifact.infra_artifact_version(_infra_init())
-
-
-def _adoptable(key: Path, url: str, want: str) -> bool:
-    """Adopt a running infra VM only if it booted from the CURRENT version and
-    its control plane is healthy. A missing/mismatched marker means a prior
-    launcher booted an older infra image — reboot rather than reuse stale code."""
-    if not key.exists():
-        return False
-    try:
-        booted = _version_file().read_text(encoding="utf-8").strip()
-    except OSError:
-        return False
-    return booted == want and _health_ok(url)
-
-
-def _record_booted_version(version: str) -> None:
-    _version_file().write_text(version + "\n", encoding="utf-8")
-
-
 # The registry "volume": a host-side ext4 file attached to the infra VM as a
 # second virtio-block device (guest /dev/vdb), mounted at the control plane's
 # DB dir. It outlives the ephemeral rootfs, so the bottle registry survives an
@@ -348,28 +309,6 @@ def _kill_pidfile() -> None:
         pass
 
 
-def _kill_infra_firecrackers(proc_root: Path = Path("/proc")) -> None:
-    """SIGKILL any firecracker VMM whose `--config-file` is this host's infra
-    config, independent of the PID file — reaps orphans it lost track of so the
-    orchestrator TAP is free to rebind. Scoped to the infra config path, so the
-    interactive pool's agent/infra VMs (other config paths) are untouched."""
-    cfg = str(_infra_dir() / "config.json")
-    for entry in proc_root.iterdir():
-        if not entry.name.isdigit():
-            continue
-        try:
-            if (entry / "comm").read_text().strip() != "firecracker":
-                continue
-            args = (entry / "cmdline").read_bytes().split(b"\0")
-        except OSError:
-            continue  # process vanished / not ours
-        if any(a.decode("utf-8", "replace") == cfg for a in args):
-            try:
-                os.kill(int(entry.name), signal.SIGKILL)
-            except (OSError, ValueError):
-                pass
-
-
 def _health_ok(url: str) -> bool:
     try:
         with urllib.request.urlopen(f"{url}/health", timeout=1.0) as resp:
@@ -494,7 +433,7 @@ BOT_BOTTLE_ROOT=/var/lib/bot-bottle python3 -m bot_bottle.orchestrator \\
 BOT_BOTTLE_GATEWAY_DAEMONS=egress,git-http,supervise \\
 BOT_BOTTLE_ORCHESTRATOR_URL=http://127.0.0.1:{CONTROL_PLANE_PORT} \\
 SUPERVISE_DB_PATH=/var/lib/bot-bottle/db/bot-bottle.db \\
-    python3 -m bot_bottle.gateway_init &
+    python3 /app/gateway_init.py &
 
 # Reap as PID 1; children are backgrounded, so `wait` blocks.
 while : ; do wait ; done

bot_bottle/backend/firecracker/publish_infra.py

@@ -11,8 +11,7 @@ The `<version>` is `infra_artifact.infra_artifact_version(...)`, the content
 hash of the rootfs inputs, so a launch host at the same code checkout resolves
 the exact artifact this produced.
 
-  python3 -m bot_bottle.backend.firecracker.publish_infra --output DIR
-  python3 -m bot_bottle.backend.firecracker.publish_infra --publish-dir DIR
+  python3 -m bot_bottle.backend.firecracker.publish_infra [--dry-run] [--force]
 
 Auth: a token with `write:package` on the target owner, from
 `BOT_BOTTLE_INFRA_ARTIFACT_TOKEN`.
@@ -25,6 +24,7 @@ import gzip
 import hashlib
 import shutil
 import sys
+import tempfile
 import urllib.error
 import urllib.request
 from pathlib import Path
@@ -131,79 +131,36 @@ def build_artifact(out_dir: Path) -> tuple[str, Path, Path]:
     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
-    # FileNotFoundError.
-    if not version_file.is_file():
-        raise SystemExit(f"incomplete artifact bundle: {root}")
-    version = version_file.read_text(encoding="utf-8").strip()
-    expected = infra_artifact.infra_artifact_version(infra_vm._infra_init())
-    if version != expected:
-        raise SystemExit(
-            f"artifact bundle version {version!r} does not match checkout {expected!r}"
-        )
-    gz = root / "rootfs.ext4.gz"
-    sha = root / "rootfs.ext4.gz.sha256"
-    if not gz.is_file() or not sha.is_file():
-        raise SystemExit(f"incomplete artifact bundle: {root}")
-    expected_sha = sha.read_text().split()[0].strip().lower()
-    if _sha256(gz) != expected_sha:
-        raise SystemExit("artifact bundle checksum mismatch")
-
-    gz_url = infra_artifact.artifact_url(version, gz.name)
-    sha_url = infra_artifact.artifact_url(version, sha.name)
-    about_url = infra_artifact.artifact_url(version, _ABOUT_NAME)
-
-    # Publishing is idempotent. If this exact complete artifact is already
-    # present, a test-only main commit is a no-op. Otherwise clear any partial
-    # upload left by an interrupted prior attempt and upload the complete set.
-    try:
-        with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp:
-            remote_sha = resp.read().decode("utf-8").split()[0].strip().lower()
-    except urllib.error.HTTPError as e:
-        if e.code != 404:
-            raise SystemExit(f"checking existing artifact failed (HTTP {e.code})")
-        remote_sha = ""
-    except urllib.error.URLError as e:
-        raise SystemExit(f"registry unreachable: {sha_url} ({e.reason})")
-    if remote_sha == expected_sha:
-        print(f"infra rootfs {version} already published")
-        return version
-
-    for url in (gz_url, sha_url, about_url):
-        _delete(url, token)
-    _put(gz_url, gz, token)
-    _put(sha_url, sha.read_bytes(), token)
-    _put(about_url, _ABOUT_TEXT.encode(), token)
-    return version
-
-
 def main(argv: list[str] | None = None) -> int:
     parser = argparse.ArgumentParser(
         prog="publish_infra", description="Build + publish the infra rootfs artifact.")
-    mode = parser.add_mutually_exclusive_group(required=True)
-    mode.add_argument("--output", type=Path,
-                      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("--dry-run", action="store_true",
+                        help="build the artifact but do not upload")
+    parser.add_argument("--force", action="store_true",
+                        help="overwrite an already-published artifact of this version")
     args = parser.parse_args(argv)
 
     _, _, token = infra_artifact._config()
-    if args.publish_dir is not None and not token:
+    if not args.dry_run and not token:
         raise SystemExit(
             "no publish token: set BOT_BOTTLE_INFRA_ARTIFACT_TOKEN to a token "
             "with write:package")
 
-    if args.output is not None:
-        args.output.mkdir(parents=True, exist_ok=True)
-        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}")
-        return 0
-
-    assert args.publish_dir is not None
-    version = _publish_bundle(args.publish_dir, token)
+    with tempfile.TemporaryDirectory(prefix="bb-publish-infra.") as tmp:
+        version, gz, sha = build_artifact(Path(tmp))
+        gz_url = infra_artifact.artifact_url(version, gz.name)
+        sha_url = infra_artifact.artifact_url(version, sha.name)
+        about_url = infra_artifact.artifact_url(version, _ABOUT_NAME)
+        if args.dry_run:
+            print(f"dry-run: would upload -> {gz_url}")
+            return 0
+        if args.force:
+            _delete(gz_url, token)
+            _delete(sha_url, token)
+            _delete(about_url, token)
+        _put(gz_url, gz, token)  # streamed from disk (hundreds of MB)
+        _put(sha_url, sha.read_bytes(), token)  # tiny, in-memory is fine
+        _put(about_url, _ABOUT_TEXT.encode(), token)  # package description
     print(f"published infra rootfs {version}")
     return 0
 

bot_bottle/backend/firecracker/util.py

@@ -368,11 +368,6 @@ mount -t devtmpfs dev /dev 2>/dev/null
 mkdir -p /dev/pts && mount -t devpts devpts /dev/pts 2>/dev/null
 mount -o remount,rw / 2>/dev/null
 
-# /tmp must be world-writable + sticky. The rootless rootfs build can land
-# it 0755/root-owned, leaving the agent (uid 1000 node) unable to create
-# scratch dirs there — git worktrees, build temp, `git init /tmp/...`, etc.
-mkdir -p /tmp && chmod 1777 /tmp
-
 # 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/contrib/claude/agent_provider.py

@@ -23,9 +23,8 @@ from ...agent_provider import (
     provider_startup_args,
 )
 from ...backend.docker import util as docker_mod
-from ...egress import CLAUDE_HOST_CREDENTIAL_TOKEN_REF, EgressRoute
+from ...egress import EgressRoute
 from ...log import die, info, warn
-from .claude_auth import claude_host_access_token
 
 
 if TYPE_CHECKING:
@@ -119,6 +118,7 @@ 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,24 +180,13 @@ 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 or forward_host_credentials) else "",
-            token_ref=cred_token_ref,
+            auth_scheme="Bearer" if auth_token else "",
+            token_ref=auth_token,
         ),)
         hidden_env_names: frozenset[str] = frozenset()
-        if auth_token or forward_host_credentials:
+        if auth_token:
             env_vars["CLAUDE_CODE_OAUTH_TOKEN"] = "egress-placeholder"
             hidden_env_names = frozenset({"CLAUDE_CODE_OAUTH_TOKEN"})
 
@@ -219,7 +208,6 @@ 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

@@ -1,114 +0,0 @@
-"""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/egress.py

@@ -30,7 +30,6 @@ 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"
 
@@ -401,7 +400,6 @@ class Egress(ABC):
         )
 
 __all__ = [
-    "CLAUDE_HOST_CREDENTIAL_TOKEN_REF",
     "CODEX_HOST_CREDENTIAL_TOKEN_REF",
     "EGRESS_HOSTNAME",
     "EGRESS_ROUTES_FILENAME",

bot_bottle/manifest_agent.py

@@ -25,9 +25,8 @@ 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 provider auth token into
-    the egress sidecar (Codex and Claude). For Codex this reads
-    `~/.codex/auth.json`; for Claude it reads `~/.claude/.credentials.json`.
+    `forward_host_credentials` forwards the host Codex auth token into
+    the egress daemon (Codex only).
     """
 
     template: str = "claude"
@@ -93,15 +92,10 @@ class ManifestAgentProvider:
                 f"is only supported for built-in templates "
                 f"({', '.join(sorted(PROVIDER_TEMPLATES))})"
             )
-        if forward_host_credentials and template not in {"codex", "claude"}:
+        if forward_host_credentials and template != "codex":
             raise ManifestError(
                 f"bottle '{bottle_name}' agent_provider.forward_host_credentials "
-                "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"
+                "is currently only supported for template 'codex'"
             )
         settings = _parse_provider_settings(bottle_name, template, d.get("settings"))
         return cls(

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

@@ -1,146 +0,0 @@
-# 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/research/agent-sandbox-landscape.md

@@ -97,8 +97,7 @@ claim that the monetization positioning leans on. See the addendum.
 
 ### agent-safehouse
 - **Source**: https://agent-safehouse.dev/ ; https://github.com/eugene1g/agent-safehouse
-- **HN launch**: [#47301085](https://news.ycombinator.com/item?id=47301085) (March 12 2026) — 823 points
-- **License**: Apache 2.0 (~1,781 stars at launch)
+- **License**: Apache 2.0 (~1,400 stars)
 - **Isolation**: macOS `sandbox-exec` (Seatbelt) profiles — kernel-level
   syscall interception, no container.
 - **Locality**: Local, macOS only.
@@ -108,16 +107,6 @@ claim that the monetization positioning leans on. See the addendum.
 - **Config**: Shell functions or custom `sandbox-exec` profile files;
   LLM-assisted profile generation supported.
 - **Network policy**: Not addressed.
-- **Notable from HN thread**: Creator acknowledged the project is "just a
-  policy-generator for `sandbox-exec` — no dependencies, no daemons, no
-  subscription; I did put in many hours to identify the minimum required
-  permissions for agents to continue working." Simon Willison noted that
-  evaluating whether a sandboxing tool actually works as intended is hard.
-  Top community sentiment: *"I honestly think that sandboxing is currently
-  THE major challenge that needs to be solved for the tech to fully realise
-  its potential."* The macOS Docker gap (Docker for Mac runs inside a Linux
-  VM, so `sandbox-exec` is the only native primitive for bare-metal macOS
-  processes) was the stated motivation.
 - **Maturity**: Active through March 2026.
 
 ### matchlock
@@ -383,7 +372,7 @@ them.
 | DX: run Claude yolo-style | One command → interactive yolo Claude (`start <agent>`, `--dangerously-skip-permissions` default) | n/a (lib demo) | Wizard + build, then run claude inside (Linux only) | One-command wrapper (`safehouse claude --dangerously-skip-permissions`) | CLI: run a cmd in a VM (not a Claude wrapper) | Hosted (`tilde exec`), not local-native | SDK code required (build the run yourself) | CLI/MCP: sandbox-as-a-tool for the agent, not a wrapper around it | SSH into a named machine, run claude there | Stand up a cluster + drive via E2B SDK | CI-oriented, not a Claude wrapper | MCP server: `claude mcp add container-use -- container-use stdio` | One command: `sbx` wraps claude with `--dangerously-skip-permissions` default | Library/wrapper, not a standalone CLI |
 | Config | JSON manifest (bottles + agents) | Programmatic refs | CLI wizard | Profile files / shell fns | CLI / SDK | DSL + CLI + SDK | SDK | CLI / SDK / MCP | TOML Smolfile | E2B-compatible SDK | cleanroom.yaml in repo | None (no policy config) | Preset levels at launch | Programmatic per-invocation (allow/deny lists) |
 | Agent-tailored policy | Yes — bottle/agent split; declarative per-role egress + credentials; composable via `extends:` | Partial — capability model scopes per-agent, but no declarative role manifest | No | Partial — per-agent profile files (Seatbelt); no egress | No | Yes — per-agent DSL RBAC (allow/deny/approve per action/repo/agent) | No | No | No | No — per-sandbox SDK config, not role-scoped | Partial — per-repo cleanroom.yaml, not per-role | No | No — network presets only | No |
-| Maturity | Active July 2026 | Research (2022+) | Early (~66 ⭐) | Active (~1.8k ⭐) | Experimental (~574 ⭐) | Private preview | YC, ~4.7k ⭐ | YC, ~6k ⭐, beta | ~3.1k ⭐ | Tencent, prod, ~10.4k ⭐ | Active (Buildkite product) | Early development | GA 2026 | Early research preview |
+| Maturity | Active July 2026 | Research (2022+) | Early (~66 ⭐) | Active (~1.4k ⭐) | Experimental (~574 ⭐) | Private preview | YC, ~4.7k ⭐ | YC, ~6k ⭐, beta | ~3.1k ⭐ | Tencent, prod, ~10.4k ⭐ | Active (Buildkite product) | Early development | GA 2026 | Early research preview |
 
 ## What's closest, what's different
 
@@ -396,9 +385,7 @@ keeping Docker only as a legacy fallback; agent-safehouse uses
 `sandbox-exec`; litterbox uses Podman + Landlock. matchlock and
 smolmachines are close on *both* the policy side (default-deny net,
 per-host allowlist) and — now that bot-bottle has moved off
-containers-by-default — the microVM isolation primitive. Note: Apple
-Container 1.0 stable shipped June 9 2026 (frozen CLI and APIs), which
-makes the macOS backend stable surface area rather than a moving target.
+containers-by-default — the microVM isolation primitive.
 
 **New closest on agent-tailored policy.** Two governance tools are the
 direct competitors on the "coarse-grained sandbox" axis. **tilde.run**

docs/research/hn-agent-safety-discourse-july-2026.md

@@ -43,18 +43,6 @@ surveyed what developers were actually deploying: "containers or YOLO"
 dominated. The honest community mood was that most teams hadn't solved
 this and were shipping anyway.
 
-The March 12 launch of **Agent Safehouse**
-([#47301085](https://news.ycombinator.com/item?id=47301085), 823 points)
-crystallised the community framing: a zero-dep `sandbox-exec` wrapper for
-macOS that attracted the top comment *"I honestly think that sandboxing is
-currently THE major challenge that needs to be solved for the tech to fully
-realise its potential."* The creator's own framing — "no dependencies, no
-daemons, no subscription; the simplicity is the feature" — and Simon
-Willison's observation that evaluating whether a sandboxing tool works as
-intended is itself hard, both prefigure the June–July shift in tone. See
-[`agent-sandbox-landscape.md`](agent-sandbox-landscape.md) for a full
-per-project breakdown.
-
 ## The June–July attack cascade
 
 Six attack patterns broke in quick succession. Together they form the

pyproject.toml

@@ -1,6 +1,6 @@
 [build-system]
 requires = ["setuptools>=68"]
-build-backend = "setuptools.build_meta"
+build-backend = "setuptools.backends.legacy:build"
 
 [project]
 name = "bot-bottle"

scripts/coverage.sh

@@ -28,7 +28,6 @@ echo "== unit ==" >&2
 
 echo "== integration (firecracker; skips docker tests) ==" >&2
 BOT_BOTTLE_BACKEND=firecracker SKIP_DOCKER_TESTS=1 \
-    BOT_BOTTLE_INFRA_ARTIFACT_DIR="${BOT_BOTTLE_CI_INFRA_ARTIFACT_DIR:-}" \
     "$PY" -m coverage run --append -m unittest discover -t . -s tests/integration
 
 echo "== combined report ==" >&2

tests/_bin.py

@@ -1,23 +0,0 @@
-"""Resolved paths to system binaries used by subprocess-based tests.
-
-NixOS and other non-FHS hosts don't populate ``/bin`` (there is no
-``/bin/sleep``), so tests that spawn real short-lived helper processes
-must resolve the binary from ``PATH`` rather than hardcoding an FHS path.
-Import the resolved constant (e.g. ``SLEEP``) instead of writing
-``/bin/sleep`` inline.
-"""
-
-from __future__ import annotations
-
-import shutil
-
-
-def resolve(name: str, fallback: str) -> str:
-    """Absolute path to ``name`` from ``PATH``; ``fallback`` on FHS hosts
-    where the binary isn't on ``PATH`` but lives at a known ``/bin`` path."""
-    return shutil.which(name) or fallback
-
-
-# Real ``sleep`` binary. ``/bin/sleep`` is absent on NixOS; resolve from
-# PATH so subprocess tests run instead of erroring with FileNotFoundError.
-SLEEP = resolve("sleep", "/bin/sleep")

tests/_docker.py

@@ -30,16 +30,3 @@ def docker_available() -> bool:
 
 def skip_unless_docker(reason: str = "docker unreachable"):
     return unittest.skipUnless(docker_available(), reason)
-
-
-def skip_unless_docker_or_firecracker(
-    reason: str = "neither Docker nor Firecracker selected",
-):
-    """Skip a backend-agnostic test unless one supported backend can run.
-
-    Firecracker does not require the host Docker daemon. The KVM coverage job
-    deliberately sets ``SKIP_DOCKER_TESTS`` to exclude Docker-only integration
-    classes while still exercising this path.
-    """
-    firecracker_selected = os.environ.get("BOT_BOTTLE_BACKEND") == "firecracker"
-    return unittest.skipUnless(firecracker_selected or docker_available(), reason)

tests/integration/test_sandbox_escape.py

@@ -31,7 +31,7 @@ from pathlib import Path
 from bot_bottle.backend import BottleSpec, get_bottle_backend
 from bot_bottle.bottle_state import cleanup_state
 from bot_bottle.manifest import ManifestIndex
-from tests._docker import skip_unless_docker_or_firecracker
+from tests._docker import skip_unless_docker
 
 
 # Secrets planted in the bottle env as literals (agents substitute via
@@ -67,7 +67,7 @@ _DUMMY_HOST_KEY = (
 )
 
 
-@skip_unless_docker_or_firecracker()
+@skip_unless_docker()
 @unittest.skipIf(
     os.environ.get("GITEA_ACTIONS") == "true"
     and os.environ.get("BOT_BOTTLE_BACKEND") != "firecracker",
@@ -91,9 +91,11 @@ class TestSandboxEscape(unittest.TestCase):
 
     @classmethod
     def setUpClass(cls) -> None:
-        # Pin Docker when BOT_BOTTLE_BACKEND is unset to preserve the
-        # Docker-backed CI path. Firecracker uses its persistent infra VM for
-        # the shared gateway and therefore does not require host Docker.
+        # Docker is always required (the agent + companion containers run under it,
+        # and VM backends still use it for the gateway); the
+        # class-level @skip_unless_docker already covers that. Pin
+        # Docker when BOT_BOTTLE_BACKEND is unset to preserve the
+        # Docker-backed CI path.
         cls._backend_name = os.environ.get("BOT_BOTTLE_BACKEND", "docker")
 
         # Throwaway static key for the git-gate fixture. It need not

tests/unit/test_agent_provider.py

@@ -9,15 +9,11 @@ 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 (
-    CLAUDE_HOST_CREDENTIAL_TOKEN_REF,
-    CODEX_HOST_CREDENTIAL_TOKEN_REF,
-)
+from bot_bottle.egress import CODEX_HOST_CREDENTIAL_TOKEN_REF
 
 
 def _jwt(exp: int) -> str:
@@ -296,67 +292,6 @@ 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_setup.py

@@ -215,18 +215,6 @@ class TestDockerSetupStatus(unittest.TestCase):
         with patch.object(dk.shutil, "which", return_value=None):
             self.assertFalse(dk._daemon_reachable())
 
-    def test_daemon_reachable_true_when_daemon_responds(self):
-        with patch.object(dk.shutil, "which", return_value="/usr/bin/docker"), \
-             patch.object(dk.subprocess, "run",
-                          return_value=subprocess.CompletedProcess([], 0)):
-            self.assertTrue(dk._daemon_reachable())
-
-    def test_daemon_reachable_false_on_timeout(self):
-        with patch.object(dk.shutil, "which", return_value="/usr/bin/docker"), \
-             patch.object(dk.subprocess, "run",
-                          side_effect=subprocess.TimeoutExpired(["docker", "info"], 5)):
-            self.assertFalse(dk._daemon_reachable())
-
     def test_status_reports_missing_docker(self):
         with patch.object(dk.shutil, "which", return_value=None):
             rc, out = _cap(dk.status)

tests/unit/test_contrib_claude_auth.py

@@ -1,186 +0,0 @@
-"""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_db_store.py

@@ -1,58 +0,0 @@
-"""Unit: DbStore._connection() context manager and is_migrated()."""
-
-from __future__ import annotations
-
-import sqlite3
-import tempfile
-import unittest
-from pathlib import Path
-
-from bot_bottle.db_store import DbStore
-from bot_bottle.migrations import TableMigrations
-
-
-def _store(tmp: Path) -> DbStore:
-    migrations = TableMigrations("test", ["CREATE TABLE items (id INTEGER PRIMARY KEY)"])
-    return DbStore(tmp / "test.db", migrations)
-
-
-class TestDbStoreIsMigrated(unittest.TestCase):
-    def test_returns_false_when_db_absent(self):
-        with tempfile.TemporaryDirectory() as d:
-            store = _store(Path(d))
-            self.assertFalse(store.is_migrated())
-
-    def test_returns_false_when_schema_versions_missing(self):
-        # DB file exists but has no schema_versions table → OperationalError → False.
-        with tempfile.TemporaryDirectory() as d:
-            store = _store(Path(d))
-            conn = sqlite3.connect(store.db_path)
-            conn.close()
-            self.assertFalse(store.is_migrated())
-
-    def test_returns_true_after_migrate(self):
-        with tempfile.TemporaryDirectory() as d:
-            store = _store(Path(d))
-            store.migrate()
-            self.assertTrue(store.is_migrated())
-
-    def test_returns_false_when_behind(self):
-        with tempfile.TemporaryDirectory() as d:
-            migrations = TableMigrations(
-                "test",
-                [
-                    "CREATE TABLE items (id INTEGER PRIMARY KEY)",
-                    "ALTER TABLE items ADD COLUMN name TEXT",
-                ],
-            )
-            store = DbStore(Path(d) / "test.db", migrations)
-            # Apply only the first migration manually.
-            conn = sqlite3.connect(store.db_path)
-            with conn:
-                TableMigrations("test", [migrations.migrations[0]]).apply(conn)
-            conn.close()
-            self.assertFalse(store.is_migrated())
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_docker_test_helpers.py

@@ -1,35 +0,0 @@
-"""Tests for integration-test backend selection helpers."""
-
-from __future__ import annotations
-
-import os
-import unittest
-from unittest.mock import patch
-
-from tests._docker import skip_unless_docker_or_firecracker
-
-
-class TestSkipUnlessDockerOrFirecracker(unittest.TestCase):
-    def test_firecracker_runs_when_docker_tests_are_disabled(self):
-        with patch.dict(
-            os.environ,
-            {"BOT_BOTTLE_BACKEND": "firecracker", "SKIP_DOCKER_TESTS": "1"},
-            clear=True,
-        ):
-            decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
-
-        self.assertFalse(getattr(decorated, "__unittest_skip__", False))
-
-    def test_non_firecracker_still_skips_when_docker_tests_are_disabled(self):
-        with patch.dict(
-            os.environ,
-            {"BOT_BOTTLE_BACKEND": "docker", "SKIP_DOCKER_TESTS": "1"},
-            clear=True,
-        ):
-            decorated = skip_unless_docker_or_firecracker()(type("Case", (), {}))
-
-        self.assertTrue(getattr(decorated, "__unittest_skip__", False))
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_firecracker_backend.py

@@ -35,12 +35,9 @@ class TestNetpoolSlots(unittest.TestCase):
     def test_slot_ip_math_31_pairs(self):
         with patch.dict(os.environ, {"BOT_BOTTLE_FC_IP_BASE": "100.64.0.0"}):
             s0, s1 = netpool.slot(0), netpool.slot(1)
-        # Iface names track netpool's (env-driven) prefix — the KVM CI runner
-        # overrides it for its isolated pool, so don't hardcode "bbfc".
-        pfx = netpool.IFACE_PREFIX
-        self.assertEqual((f"{pfx}0", "100.64.0.0", "100.64.0.1"),
+        self.assertEqual(("bbfc0", "100.64.0.0", "100.64.0.1"),
                          (s0.iface, s0.host_ip, s0.guest_ip))
-        self.assertEqual((f"{pfx}1", "100.64.0.2", "100.64.0.3"),
+        self.assertEqual(("bbfc1", "100.64.0.2", "100.64.0.3"),
                          (s1.iface, s1.host_ip, s1.guest_ip))
 
     def test_guest_cidr_is_31(self):
@@ -183,14 +180,11 @@ class TestNetpoolOverlap(unittest.TestCase):
         self.assertEqual("tailscale0", conflicts[0].dev)
 
     def test_ignores_own_taps_and_default(self):
-        # The "own tap" route uses netpool's (env-driven) iface name, so the
-        # test still exercises the self-ignore path on the KVM CI runner, whose
-        # BOT_BOTTLE_FC_IFACE_PREFIX differs from the default.
         with patch.dict(os.environ, {"BOT_BOTTLE_FC_IP_BASE": "10.243.0.0",
                                      "BOT_BOTTLE_FC_POOL_SIZE": "8"}), \
              self._routes([
                  {"dst": "default", "dev": "enp4s0"},
-                 {"dst": "10.243.0.0/31", "dev": netpool.slot(0).iface},
+                 {"dst": "10.243.0.0/31", "dev": "bbfc0"},
                  {"dst": "192.168.1.0/24", "dev": "enp4s0"},
              ]):
             self.assertEqual([], netpool.overlapping_routes())
@@ -209,7 +203,7 @@ class TestNetpoolAllocation(unittest.TestCase):
             # over (and here, exhaust the pool).
             slot, lock = netpool.allocate("first")
             self.addCleanup(lock.close)
-            self.assertEqual(netpool.slot(0).iface, slot.iface)
+            self.assertEqual("bbfc0", slot.iface)
             with patch.object(netpool, "die",
                               side_effect=SystemExit("exhausted")):
                 with self.assertRaises(SystemExit):
@@ -329,14 +323,9 @@ class TestNetpoolDefaultsSingleSource(unittest.TestCase):
         with patch.dict(os.environ, {}, clear=True):
             self.assertEqual(int(d["BOT_BOTTLE_FC_POOL_SIZE"]), netpool.pool_size())
             self.assertEqual(d["BOT_BOTTLE_FC_IP_BASE"], netpool.ip_base())
-        # The module's *defaults* come from the same shared file. Assert the
-        # parsed defaults (not IFACE_PREFIX/NFT_TABLE, which layer a live env
-        # override on top — the KVM CI runner sets those for its isolated pool,
-        # which would otherwise mask this single-source check).
-        self.assertEqual(d["BOT_BOTTLE_FC_IFACE_PREFIX"],
-                         netpool._DEFAULTS["BOT_BOTTLE_FC_IFACE_PREFIX"])
-        self.assertEqual(d["BOT_BOTTLE_FC_NFT_TABLE"],
-                         netpool._DEFAULTS["BOT_BOTTLE_FC_NFT_TABLE"])
+        # Module constants resolve through the same shared file.
+        self.assertEqual(d["BOT_BOTTLE_FC_IFACE_PREFIX"], netpool.IFACE_PREFIX)
+        self.assertEqual(d["BOT_BOTTLE_FC_NFT_TABLE"], netpool.NFT_TABLE)
 
     def test_env_var_overrides_the_shared_default(self):
         with patch.dict(os.environ, {"BOT_BOTTLE_FC_IP_BASE": "10.99.0.0"}):

tests/unit/test_firecracker_helpers.py

@@ -63,12 +63,9 @@ class TestNetpoolProbes(unittest.TestCase):
         self.assertEqual(2, ok.call_count)
 
     def test_missing_taps(self):
-        # Derive the expected iface from netpool's (env-driven) config rather
-        # than hardcoding "bbfc1": the KVM CI runner sets BOT_BOTTLE_FC_* for
-        # its isolated pool, so the prefix there is not the default.
         with patch.dict("os.environ", {"BOT_BOTTLE_FC_POOL_SIZE": "2"}), \
              patch.object(netpool, "tap_present", side_effect=[True, False]):
-            self.assertEqual([netpool.slot(1).iface], netpool.missing_taps())
+            self.assertEqual(["bbfc1"], netpool.missing_taps())
 
     def test_orch_slot_is_top_of_ip_base_16(self):
         # Dedicated orchestrator link: /31 at the top of the IP_BASE /16,

tests/unit/test_firecracker_image_builder.py

@@ -24,7 +24,7 @@ class TestBuildAgentRootfsDir(unittest.TestCase):
         self.addCleanup(self._tmp.cleanup)
 
     def test_cache_hit_skips_rebuild(self):
-        digest = image_builder._rootfs_digest(self.dockerfile)
+        digest = image_builder._dockerfile_hash(self.dockerfile)
         base = self.cache / "rootfs" / f"agent-{digest}"
         base.mkdir(parents=True)
         (base / ".bb-ready").write_text("ok\n")
@@ -55,17 +55,6 @@ class TestBuildAgentRootfsDir(unittest.TestCase):
             image_builder._dockerfile_hash(other),
         )
 
-    def test_rootfs_digest_tracks_dockerfile_and_init(self):
-        # Same Dockerfile, different injected init -> different rootfs key, so
-        # an init fix (e.g. /tmp perms) rebuilds instead of reusing a stale
-        # rootfs; different Dockerfiles also differ.
-        base = image_builder._rootfs_digest(self.dockerfile)
-        with patch.object(image_builder.util, "_GUEST_INIT", "#!/bin/sh\n# changed\n"):
-            self.assertNotEqual(base, image_builder._rootfs_digest(self.dockerfile))
-        other = self.cache / "Dockerfile2"
-        other.write_text("FROM python:3.12-slim\n")
-        self.assertNotEqual(base, image_builder._rootfs_digest(other))
-
 
 class TestSmokeTest(unittest.TestCase):
     def test_empty_argv_is_noop(self):

tests/unit/test_firecracker_infra_vm.py

@@ -38,9 +38,7 @@ class TestBuildInfraRootfs(unittest.TestCase):
         # and exports PATH so gateway_init's subprocess daemons find python3.
         init = build.call_args.kwargs["init_script"]
         self.assertIn("bot_bottle.orchestrator", init)
-        # Gateway launches via the installed package (there is no
-        # /app/gateway_init.py file since the daemons moved into bot_bottle).
-        self.assertIn("bot_bottle.gateway_init", init)
+        self.assertIn("gateway_init.py", init)
         self.assertIn("export PATH=", init)
         # Persistent registry volume mounted at the DB dir before the CP starts.
         self.assertIn("/dev/vdb", init)
@@ -100,11 +98,7 @@ class TestRegistryVolume(unittest.TestCase):
 class TestEnsureBuilt(unittest.TestCase):
     def test_default_pulls_artifact_without_docker(self):
         # PRD 0069 Stage 2: the launch host pulls the prebuilt rootfs; no Docker.
-        # Pin BOT_BOTTLE_INFRA_BUILD off: the coverage CI job exports it =local
-        # for the integration suite, and that ambient value would otherwise send
-        # this default-path test down the local Docker-build branch.
-        with patch.dict(os.environ, {"BOT_BOTTLE_INFRA_BUILD": ""}), \
-                patch.object(infra_vm.docker_mod, "build_image") as build, \
+        with patch.object(infra_vm.docker_mod, "build_image") as build, \
                 patch.object(infra_vm.infra_artifact, "ensure_artifact_gz") as pull:
             infra_vm.ensure_built()
         build.assert_not_called()
@@ -144,58 +138,27 @@ class TestWaitForHealth(unittest.TestCase):
 
 
 class TestEnsureRunningSingleton(unittest.TestCase):
-    def test_adopts_when_healthy_and_version_matches(self):
-        # Healthy control plane + existing key + matching version marker
-        # -> adopt (no boot), vm=None.
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = Path(td)
-            (d / "id_ed25519").write_text("k")
-            (d / "booted-version").write_text("v-current\n")
-            with patch.object(infra_vm, "_infra_dir", return_value=d), \
-                 patch.object(infra_vm, "_expected_version", return_value="v-current"), \
-                 patch.object(infra_vm, "_health_ok", return_value=True), \
-                 patch.object(infra_vm, "boot") as boot:
-                infra = infra_vm.ensure_running()
+    def test_adopts_when_healthy(self):
+        # A healthy control plane + existing key -> adopt (no boot), vm=None.
+        with patch.object(infra_vm, "_health_ok", return_value=True), \
+             patch.object(infra_vm, "_infra_dir") as d, \
+             patch.object(infra_vm, "boot") as boot:
+            keydir = MagicMock()
+            (keydir / "id_ed25519").exists.return_value = True
+            d.return_value = keydir
+            infra = infra_vm.ensure_running()
         boot.assert_not_called()
         self.assertIsNone(infra.vm)
 
-    def test_reboots_when_version_stale(self):
-        # Healthy control plane but the running VM booted an OLDER image
-        # (marker mismatch) -> reboot rather than adopt stale code.
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = Path(td)
-            (d / "id_ed25519").write_text("k")
-            (d / "booted-version").write_text("v-old\n")
-            with patch.object(infra_vm, "_infra_dir", return_value=d), \
-                 patch.object(infra_vm, "_expected_version", return_value="v-current"), \
-                 patch.object(infra_vm, "_health_ok", return_value=True), \
-                 patch.object(infra_vm, "stop") as stop, \
-                 patch.object(infra_vm, "ensure_built"), \
-                 patch.object(infra_vm, "wait_for_health"), \
-                 patch.object(infra_vm, "boot") as boot:
-                boot.return_value = infra_vm.InfraVm(
-                    guest_ip="10.243.255.1", private_key=Path("/k"), vm=MagicMock())
-                infra_vm.ensure_running()
-            stop.assert_called_once()      # dislodge the outdated VM
-            boot.assert_called_once()
-            # The fresh boot records the current version for the next launcher.
-            self.assertEqual("v-current\n", (d / "booted-version").read_text())
-
     def test_boots_when_unhealthy(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            with patch.object(infra_vm, "_infra_dir", return_value=Path(td)), \
-                 patch.object(infra_vm, "_expected_version", return_value="v-current"), \
-                 patch.object(infra_vm, "_health_ok", return_value=False), \
-                 patch.object(infra_vm, "stop") as stop, \
-                 patch.object(infra_vm, "ensure_built") as built, \
-                 patch.object(infra_vm, "boot") as boot, \
-                 patch.object(infra_vm, "wait_for_health") as wait:
-                boot.return_value = infra_vm.InfraVm(
-                    guest_ip="10.243.255.1", private_key=Path("/k"), vm=MagicMock())
-                infra_vm.ensure_running()
+        with patch.object(infra_vm, "_health_ok", return_value=False), \
+             patch.object(infra_vm, "stop") as stop, \
+             patch.object(infra_vm, "ensure_built") as built, \
+             patch.object(infra_vm, "boot") as boot, \
+             patch.object(infra_vm, "wait_for_health") as wait:
+            boot.return_value = infra_vm.InfraVm(
+                guest_ip="10.243.255.1", private_key=Path("/k"), vm=MagicMock())
+            infra_vm.ensure_running()
         stop.assert_called_once()      # clear a stale VM first
         built.assert_called_once()
         boot.assert_called_once()
@@ -222,74 +185,5 @@ class TestKillPidfile(unittest.TestCase):
             kill.assert_not_called()
 
 
-class TestAdoptable(unittest.TestCase):
-    def _dir(self, td: str, *, key: bool = True, version: str | None = None) -> Path:
-        d = Path(td)
-        if key:
-            (d / "id_ed25519").write_text("k")
-        if version is not None:
-            (d / "booted-version").write_text(version + "\n")
-        return d
-
-    def test_true_when_key_version_and_health(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = self._dir(td, version="v1")
-            with patch.object(infra_vm, "_infra_dir", return_value=d), \
-                 patch.object(infra_vm, "_health_ok", return_value=True):
-                self.assertTrue(infra_vm._adoptable(d / "id_ed25519", "u", "v1"))
-
-    def test_false_when_key_missing(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = self._dir(td, key=False, version="v1")
-            with patch.object(infra_vm, "_infra_dir", return_value=d):
-                self.assertFalse(infra_vm._adoptable(d / "id_ed25519", "u", "v1"))
-
-    def test_false_when_no_version_marker(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = self._dir(td)  # key present, no booted-version
-            with patch.object(infra_vm, "_infra_dir", return_value=d):
-                self.assertFalse(infra_vm._adoptable(d / "id_ed25519", "u", "v1"))
-
-    def test_false_when_version_mismatch(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td:
-            d = self._dir(td, version="v-old")
-            with patch.object(infra_vm, "_infra_dir", return_value=d), \
-                 patch.object(infra_vm, "_health_ok", return_value=True):
-                self.assertFalse(infra_vm._adoptable(d / "id_ed25519", "u", "v1"))
-
-
-class TestKillInfraFirecrackers(unittest.TestCase):
-    def _fake_proc(self, root: Path, pid: int, comm: str, cmdline: list[str]) -> None:
-        p = root / str(pid)
-        p.mkdir()
-        (p / "comm").write_text(comm + "\n")
-        (p / "cmdline").write_bytes(b"\0".join(a.encode() for a in cmdline) + b"\0")
-
-    def test_kills_only_matching_infra_firecracker(self):
-        import tempfile
-        with tempfile.TemporaryDirectory() as td, \
-             tempfile.TemporaryDirectory() as proc:
-            infra_dir = Path(td)
-            cfg = str(infra_dir / "config.json")
-            root = Path(proc)
-            # target: firecracker bound to the infra config -> killed
-            self._fake_proc(root, 111, "firecracker",
-                            ["firecracker", "--no-api", "--config-file", cfg])
-            # a firecracker for a different (interactive) VM -> spared
-            self._fake_proc(root, 222, "firecracker",
-                            ["firecracker", "--config-file", "/home/u/other.json"])
-            # a non-firecracker process on the same config path -> spared
-            self._fake_proc(root, 333, "python3", ["python3", cfg])
-            (root / "not-a-pid").mkdir()
-            with patch.object(infra_vm, "_infra_dir", return_value=infra_dir), \
-                 patch.object(infra_vm.os, "kill") as kill:
-                infra_vm._kill_infra_firecrackers(proc_root=root)
-            kill.assert_called_once_with(111, infra_vm.signal.SIGKILL)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_gateway_init.py

@@ -2,9 +2,8 @@
 
 Tests both the helper functions in `bot_bottle.gateway_init`
 and the supervisor's end-to-end signal / exit-code behavior. The
-end-to-end tests use real subprocesses (`sleep`, `/bin/sh -c '...'`) —
-short-lived, no docker required — so they run under `tests/unit/`
-rather than `tests/integration/`."""
+end-to-end tests use real subprocesses — short-lived, no docker required
+— so they run under `tests/unit/` rather than `tests/integration/`."""
 
 from __future__ import annotations
 
@@ -15,7 +14,6 @@ import sys
 import time
 import unittest
 import warnings
-from pathlib import Path
 from unittest.mock import patch
 
 from bot_bottle.gateway_init import (
@@ -25,7 +23,11 @@ from bot_bottle.gateway_init import (
     _env_for_daemon,
     _selected_daemons,
 )
-from tests._bin import SLEEP
+
+# /bin/sleep does not exist on FHS-free systems (e.g. NixOS). Use a
+# portable Python one-liner so supervisor tests run on any platform.
+_SLEEP_30 = (sys.executable, "-c", "import time; time.sleep(30)")
+_SLEEP_60 = (sys.executable, "-c", "import time; time.sleep(60)")
 
 
 class TestEnvForDaemon(unittest.TestCase):
@@ -183,7 +185,7 @@ class TestSupervisor(unittest.TestCase):
         # up and the supervisor never set shutdown_at.
         specs = [
             _DaemonSpec("crasher", ("/bin/sh", "-c", "exit 1")),
-            _DaemonSpec("longrun", (SLEEP, "30")),
+            _DaemonSpec("longrun", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -215,7 +217,7 @@ class TestSupervisor(unittest.TestCase):
         # signal-killed longrun's negative returncode.
         specs = [
             _DaemonSpec("crasher", ("/bin/sh", "-c", "exit 1")),
-            _DaemonSpec("longrun", (SLEEP, "30")),
+            _DaemonSpec("longrun", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -260,7 +262,7 @@ class TestSupervisor(unittest.TestCase):
         )
         specs = [
             _DaemonSpec("egress", sighup_marker),
-            _DaemonSpec("other",  (SLEEP, "30")),
+            _DaemonSpec("other",  _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -283,7 +285,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_forward_signal_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", (SLEEP, "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         delivered = sup.forward_signal(signal.SIGHUP, "ghost")
@@ -295,8 +297,8 @@ class TestSupervisor(unittest.TestCase):
         # Restart one daemon; the other (supervise, the MCP server
         # in production) must remain untouched.
         specs = [
-            _DaemonSpec("git-gate", (SLEEP, "30")),
-            _DaemonSpec("supervise", (SLEEP, "30")),
+            _DaemonSpec("git-gate", _SLEEP_30),
+            _DaemonSpec("supervise", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -320,8 +322,8 @@ class TestSupervisor(unittest.TestCase):
 
     def test_request_restart_is_drained_by_tick(self):
         specs = [
-            _DaemonSpec("git-gate", (SLEEP, "30")),
-            _DaemonSpec("supervise", (SLEEP, "30")),
+            _DaemonSpec("git-gate", _SLEEP_30),
+            _DaemonSpec("supervise", _SLEEP_30),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -344,7 +346,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_repeated_restart_requests_coalesce(self):
-        specs = [_DaemonSpec("git-gate", (SLEEP, "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -367,7 +369,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_request_restart_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", (SLEEP, "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         ok = sup.request_restart("ghost")
@@ -377,7 +379,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_restart_unknown_daemon_no_op(self):
-        specs = [_DaemonSpec("a", (SLEEP, "30"))]
+        specs = [_DaemonSpec("a", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         ok = sup.restart_daemon("ghost")
@@ -386,7 +388,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_restart_during_shutdown_is_no_op(self):
-        specs = [_DaemonSpec("git-gate", (SLEEP, "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         sup.request_shutdown(reason="test")
@@ -396,7 +398,7 @@ class TestSupervisor(unittest.TestCase):
         self._drive(sup)
 
     def test_pending_restart_dropped_during_shutdown(self):
-        specs = [_DaemonSpec("git-gate", (SLEEP, "30"))]
+        specs = [_DaemonSpec("git-gate", _SLEEP_30)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -414,8 +416,8 @@ class TestSupervisor(unittest.TestCase):
         # both should receive SIGTERM and exit. Signal-only
         # shutdown clamps to a zero supervisor exit code.
         specs = [
-            _DaemonSpec("a", (SLEEP, "60")),
-            _DaemonSpec("b", (SLEEP, "60")),
+            _DaemonSpec("a", _SLEEP_60),
+            _DaemonSpec("b", _SLEEP_60),
         ]
         sup = _Supervisor(specs)
         sup.start_all()
@@ -450,7 +452,7 @@ class TestSupervisor(unittest.TestCase):
         self.assertEqual(0, rc)
 
     def test_idempotent_shutdown_requests(self):
-        specs = [_DaemonSpec("a", (SLEEP, "60"))]
+        specs = [_DaemonSpec("a", _SLEEP_60)]
         sup = _Supervisor(specs)
         sup.start_all()
         time.sleep(0.1)
@@ -466,28 +468,22 @@ class TestSupervisor(unittest.TestCase):
 
 class TestMainEndToEnd(unittest.TestCase):
     """Run gateway_init.py as a real subprocess to cover the
-    signal-handler installation path. Skipped on platforms
-    without /bin/sleep + /bin/sh."""
-
-    @classmethod
-    def setUpClass(cls):
-        for p in ("/bin/sh", SLEEP):
-            if not Path(p).exists():
-                raise unittest.SkipTest(f"missing {p}")
+    signal-handler installation path."""
 
     def _run(self, daemons_csv: str, send_signal: int | None,
              wait_before_signal: float = 0.4,
              overall_timeout: float = 6.0) -> tuple[int, str]:
         """Spawn gateway_init.main() in a child process with the
-        DAEMONS list patched to harmless `sleep 30` commands.
+        DAEMONS list patched to harmless long-sleep commands.
         Returns (returncode, captured stdout)."""
 
         helper = (
             "import os, runpy, sys\n"
             "from bot_bottle import gateway_init as si\n"
+            "sleep_cmd = (sys.executable, '-c', 'import time; time.sleep(30)')\n"
             "si._DAEMONS = (\n"
-            f"  si._DaemonSpec('alpha', ({SLEEP!r},'30')),\n"
-            f"  si._DaemonSpec('beta',  ({SLEEP!r},'30')),\n"
+            "  si._DaemonSpec('alpha', sleep_cmd),\n"
+            "  si._DaemonSpec('beta',  sleep_cmd),\n"
             ")\n"
             "sys.exit(si.main([]))\n"
         )

tests/unit/test_infra_artifact.py

@@ -50,12 +50,7 @@ class _CacheMixin(unittest.TestCase):
         self._env = mock.patch.dict(
             os.environ,
             {"BOT_BOTTLE_FC_CACHE": self._tmp.name,
-             "BOT_BOTTLE_INFRA_ARTIFACT_TOKEN": "",
-             # Pin the candidate-dir override off: the coverage CI job exports a
-             # candidate dir for the integration suite, and an ambient value
-             # would send these registry-pull tests down the local-bundle path.
-             # Cases that exercise the candidate path set it explicitly.
-             "BOT_BOTTLE_INFRA_ARTIFACT_DIR": ""},
+             "BOT_BOTTLE_INFRA_ARTIFACT_TOKEN": ""},
             clear=False,
         )
         self._env.start()
@@ -98,31 +93,6 @@ class TestVersionInputs(unittest.TestCase):
         (pkg / "netpool.defaults.env").write_text("FOO=1\n")
         for name in ("Dockerfile.orchestrator", "Dockerfile.gateway", "Dockerfile.infra"):
             (root / name).write_text(f"FROM scratch  # {name}\n")
-        (root / "pyproject.toml").write_text("[project]\nname = 'bot-bottle'\n")
-
-    def test_pyproject_toml_change_bumps_version(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._fake_repo(root)
-            before = ia.infra_artifact_version("init", repo_root=root)
-            (root / "pyproject.toml").write_text(
-                "[project]\nname = 'bot-bottle'\ndependencies = ['httpx']\n")
-            after = ia.infra_artifact_version("init", repo_root=root)
-        self.assertNotEqual(before, after)
-
-    def test_dropbear_change_bumps_version(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._fake_repo(root)
-            dropbear = root / "dropbear"
-            dropbear.write_bytes(b"dropbear-v1")
-            with mock.patch.dict(os.environ, {
-                "BOT_BOTTLE_FC_DROPBEAR": str(dropbear),
-            }):
-                before = ia.infra_artifact_version("init", repo_root=root)
-                dropbear.write_bytes(b"dropbear-v2")
-                after = ia.infra_artifact_version("init", repo_root=root)
-        self.assertNotEqual(before, after)
 
     def test_non_python_file_change_bumps_version(self) -> None:
         with tempfile.TemporaryDirectory() as d:
@@ -148,58 +118,6 @@ class TestVersionInputs(unittest.TestCase):
 
 
 class TestEnsureArtifact(_CacheMixin):
-    def test_uses_verified_ci_candidate_without_network(self) -> None:
-        version = "deadbeef00000000"
-        gz = _gz(b"candidate ext4")
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            (root / "version.txt").write_text(version + "\n")
-            (root / "rootfs.ext4.gz").write_bytes(gz)
-            digest = hashlib.sha256(gz).hexdigest()
-            (root / "rootfs.ext4.gz.sha256").write_text(
-                f"{digest}  rootfs.ext4.gz\n")
-            with mock.patch.dict(os.environ, {
-                "BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
-            }), mock.patch.object(ia.urllib.request, "urlopen") as net:
-                path = ia.ensure_artifact_gz(version)
-            self.assertEqual(root / "rootfs.ext4.gz", path)
-            net.assert_not_called()
-
-    def test_rejects_candidate_for_another_version(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            (root / "version.txt").write_text("wrong\n")
-            with mock.patch.dict(os.environ, {
-                "BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
-            }):
-                with self.assertRaises(Die) as ctx:
-                    ia.ensure_artifact_gz("expected")
-            self.assertIn("version mismatch", str(ctx.exception.message))
-
-    def test_rejects_incomplete_candidate(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            (root / "version.txt").write_text("v1\n")
-            with mock.patch.dict(os.environ, {
-                "BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
-            }):
-                with self.assertRaises(Die) as ctx:
-                    ia.ensure_artifact_gz("v1")
-            self.assertIn("incomplete", str(ctx.exception.message))
-
-    def test_rejects_candidate_checksum_mismatch(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            (root / "version.txt").write_text("v1\n")
-            (root / "rootfs.ext4.gz").write_bytes(b"bad")
-            (root / "rootfs.ext4.gz.sha256").write_text("0" * 64 + "  rootfs.ext4.gz\n")
-            with mock.patch.dict(os.environ, {
-                "BOT_BOTTLE_INFRA_ARTIFACT_DIR": str(root),
-            }):
-                with self.assertRaises(Die) as ctx:
-                    ia.ensure_artifact_gz("v1")
-            self.assertIn("checksum mismatch", str(ctx.exception.message))
-
     def test_downloads_verifies_and_caches(self) -> None:
         version = "deadbeef00000000"
         gz = _gz(b"fake ext4 bytes")

tests/unit/test_manifest_egress.py

@@ -80,19 +80,11 @@ class TestAgentProviderHostCredentials(unittest.TestCase):
                 "forward_host_credentials": "yes",
             })
 
-    def test_forward_host_credentials_allowed_for_claude(self):
-        b = _provider_config_bottle({
-            "template": "claude",
-            "forward_host_credentials": True,
-        })
-        self.assertTrue(b.agent_provider.forward_host_credentials)
-
-    def test_forward_host_credentials_and_auth_token_rejected_together(self):
+    def test_forward_host_credentials_rejected_for_claude(self):
         with self.assertRaises(ManifestError):
             _provider_config_bottle({
                 "template": "claude",
                 "forward_host_credentials": True,
-                "auth_token": "SOME_TOKEN",
             })
 
     def test_auth_token_defaults_empty(self):

tests/unit/test_manifest_validation.py

@@ -86,22 +86,10 @@ class TestAgentProviderValidation(unittest.TestCase):
                 "b", {"forward_host_credentials": True, "template": "weird"}
             )
 
-    def test_forward_creds_pi_template_rejected(self) -> None:
+    def test_forward_creds_non_codex_template(self) -> None:
         with self.assertRaises(ManifestError):
             ManifestAgentProvider.from_dict(
-                "b", {"forward_host_credentials": True, "template": "pi"}
-            )
-
-    def test_forward_creds_claude_allowed(self) -> None:
-        p = ManifestAgentProvider.from_dict(
-            "b", {"forward_host_credentials": True, "template": "claude"}
-        )
-        self.assertTrue(p.forward_host_credentials)
-
-    def test_forward_creds_and_auth_token_rejected(self) -> None:
-        with self.assertRaises(ManifestError):
-            ManifestAgentProvider.from_dict(
-                "b", {"forward_host_credentials": True, "auth_token": "T", "template": "claude"}
+                "b", {"forward_host_credentials": True, "template": "claude"}
             )
 
     def test_valid_claude_auth_token(self) -> None:

tests/unit/test_publish_infra.py

@@ -6,11 +6,8 @@ read it into memory. Network is mocked; no Docker, no real build.
 
 from __future__ import annotations
 
-import hashlib
-from email.message import Message
 import tempfile
 import unittest
-import urllib.error
 import urllib.request
 from pathlib import Path
 from unittest import mock
@@ -61,99 +58,5 @@ class TestPut(unittest.TestCase):
         self.assertEqual(b"abc123  rootfs\n", captured[0].data)
 
 
-class TestPublishBundle(unittest.TestCase):
-    def _bundle(self, root: Path, version: str) -> None:
-        payload = b"candidate"
-        (root / "version.txt").write_text(version + "\n")
-        (root / "rootfs.ext4.gz").write_bytes(payload)
-        digest = hashlib.sha256(payload).hexdigest()
-        (root / "rootfs.ext4.gz.sha256").write_text(
-            f"{digest}  rootfs.ext4.gz\n")
-
-    def test_existing_identical_artifact_is_success(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._bundle(root, "v1")
-            sha = (root / "rootfs.ext4.gz.sha256").read_bytes()
-            response = mock.MagicMock()
-            response.__enter__.return_value.read.return_value = sha
-            with mock.patch.object(
-                pub.infra_artifact, "infra_artifact_version", return_value="v1"
-            ), mock.patch.object(
-                pub.urllib.request, "urlopen", return_value=response
-            ), mock.patch.object(pub, "_put") as put:
-                self.assertEqual("v1", pub._publish_bundle(root, "token"))
-            put.assert_not_called()
-
-    def test_partial_artifact_is_replaced(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._bundle(root, "v1")
-            missing = urllib.error.HTTPError("u", 404, "missing", Message(), None)
-            with mock.patch.object(
-                pub.infra_artifact, "infra_artifact_version", return_value="v1"
-            ), mock.patch.object(
-                pub.urllib.request, "urlopen", side_effect=missing
-            ), mock.patch.object(pub, "_delete") as delete, \
-                    mock.patch.object(pub, "_put") as put:
-                pub._publish_bundle(root, "token")
-            self.assertEqual(3, delete.call_count)
-            self.assertEqual(3, put.call_count)
-
-    def test_rejects_bundle_for_different_checkout(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._bundle(root, "old")
-            with mock.patch.object(
-                pub.infra_artifact, "infra_artifact_version", return_value="new"
-            ):
-                with self.assertRaises(SystemExit) as ctx:
-                    pub._publish_bundle(root, "token")
-            self.assertIn("does not match checkout", str(ctx.exception))
-
-    def test_rejects_bad_bundle_checksum(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._bundle(root, "v1")
-            (root / "rootfs.ext4.gz").write_bytes(b"tampered")
-            with mock.patch.object(
-                pub.infra_artifact, "infra_artifact_version", return_value="v1"
-            ):
-                with self.assertRaises(SystemExit) as ctx:
-                    pub._publish_bundle(root, "token")
-            self.assertIn("checksum mismatch", str(ctx.exception))
-
-    def test_registry_lookup_failure_is_reported(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d)
-            self._bundle(root, "v1")
-            failure = urllib.error.URLError("offline")
-            with mock.patch.object(
-                pub.infra_artifact, "infra_artifact_version", return_value="v1"
-            ), mock.patch.object(pub.urllib.request, "urlopen", side_effect=failure):
-                with self.assertRaises(SystemExit) as ctx:
-                    pub._publish_bundle(root, "token")
-            self.assertIn("registry unreachable", str(ctx.exception))
-
-
-class TestMain(unittest.TestCase):
-    def test_output_builds_candidate_and_records_version(self) -> None:
-        with tempfile.TemporaryDirectory() as d:
-            root = Path(d) / "candidate"
-            with mock.patch.object(
-                pub, "build_artifact", return_value=("v1", root / "g", root / "s")
-            ) as build:
-                self.assertEqual(0, pub.main(["--output", str(root)]))
-            build.assert_called_once_with(root)
-            self.assertEqual("v1\n", (root / "version.txt").read_text())
-
-    def test_publish_dir_publishes_existing_candidate(self) -> None:
-        with tempfile.TemporaryDirectory() as d, \
-                mock.patch.object(pub.infra_artifact, "_config", return_value=("", "", "t")), \
-                mock.patch.object(pub, "_publish_bundle", return_value="v1") as publish:
-            self.assertEqual(0, pub.main(["--publish-dir", d]))
-            publish.assert_called_once_with(Path(d), "t")
-
-
 if __name__ == "__main__":
     unittest.main()