fix: enforce cleanup and secret integrity

bot_bottle/backend/base.py

@@ -172,6 +172,10 @@ class BottleCleanupPlan(ABC):
         """True iff there is nothing to clean up; the CLI uses this to
         short-circuit before showing the y/N."""
 
+    @abstractmethod
+    def intersect(self, current: "BottleCleanupPlan") -> "BottleCleanupPlan":
+        """Resources both displayed to the operator and currently removable."""
+
 
 @dataclass(frozen=True)
 class ExecResult:

bot_bottle/backend/cleanup_control.py

@@ -0,0 +1,48 @@
+"""Shared destructive-cleanup execution and failure accounting."""
+
+from __future__ import annotations
+
+import shutil
+import subprocess
+from collections.abc import Sequence
+from pathlib import Path
+
+
+class CleanupError(RuntimeError):
+    """One or more approved cleanup mutations did not complete."""
+
+
+class CleanupFailures:
+    """Attempt every approved mutation, then fail with complete diagnostics."""
+
+    def __init__(self) -> None:
+        self._messages: list[str] = []
+
+    def run(self, argv: Sequence[str], description: str) -> None:
+        try:
+            result = subprocess.run(
+                list(argv), capture_output=True, text=True, check=False,
+            )
+        except OSError as exc:
+            self._messages.append(f"{description}: {exc}")
+            return
+        if result.returncode != 0:
+            detail = (result.stderr or result.stdout).strip()
+            self._messages.append(
+                f"{description}: {detail or f'exit {result.returncode}'}"
+            )
+
+    def remove_tree(self, path: Path, description: str) -> None:
+        try:
+            shutil.rmtree(path)
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            self._messages.append(f"{description}: {exc}")
+
+    def raise_if_any(self) -> None:
+        if self._messages:
+            raise CleanupError("; ".join(self._messages))
+
+
+__all__ = ["CleanupError", "CleanupFailures"]

bot_bottle/backend/docker/bottle_cleanup_plan.py

@@ -46,6 +46,22 @@ class DockerBottleCleanupPlan(BottleCleanupPlan):
             and not self.orphan_state_dirs
         )
 
+    def intersect(self, current: BottleCleanupPlan) -> "DockerBottleCleanupPlan":
+        if not isinstance(current, DockerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return DockerBottleCleanupPlan(
+            projects=tuple(x for x in self.projects if x in current.projects),
+            stray_containers=tuple(
+                x for x in self.stray_containers if x in current.stray_containers
+            ),
+            stray_networks=tuple(
+                x for x in self.stray_networks if x in current.stray_networks
+            ),
+            orphan_state_dirs=tuple(
+                x for x in self.orphan_state_dirs if x in current.orphan_state_dirs
+            ),
+        )
+
     def print(self) -> None:
         print(file=sys.stderr)
         for name in self.projects:

bot_bottle/backend/docker/cleanup.py

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

bot_bottle/backend/firecracker/bottle_cleanup_plan.py

@@ -27,3 +27,11 @@ class FirecrackerBottleCleanupPlan(BottleCleanupPlan):
     @property
     def empty(self) -> bool:
         return not (self.vm_pids or self.run_dirs)
+
+    def intersect(self, current: BottleCleanupPlan) -> "FirecrackerBottleCleanupPlan":
+        if not isinstance(current, FirecrackerBottleCleanupPlan):
+            raise TypeError("cleanup plans must have the same backend type")
+        return FirecrackerBottleCleanupPlan(
+            vm_pids=tuple(x for x in self.vm_pids if x in current.vm_pids),
+            run_dirs=tuple(x for x in self.run_dirs if x in current.run_dirs),
+        )

bot_bottle/backend/firecracker/cleanup.py

@@ -11,10 +11,9 @@ Reaps *orphans* only — resources with no live VM behind them:
     — a VMM left lingering after its dir was removed.
 
 A run dir with a *live* firecracker process is a running bottle and is
-left strictly alone: it is neither killed nor removed. (The backend's
-`enumerate_active` registry is still a stub — #354 — so a live process
-is the only reliable "this bottle is in use" signal we have. Once the
-registry lands, registry-orphaned-but-running VMs can be reaped too.)
+left strictly alone: it is neither killed nor removed. Active-agent
+enumeration uses this same process snapshot, so cleanup and generic
+backend consumers agree about which bottles are running.
 
 TAP slots free themselves (the flock drops when the launcher exits), so
 there is nothing to reclaim there.
@@ -22,15 +21,16 @@ there is nothing to reclaim there.
 
 from __future__ import annotations
 
+from collections.abc import Sequence
 import os
-import shutil
 import signal
 import subprocess
 from pathlib import Path
 
 from ...log import info
 from .. import EnumerationError
-from . import util
+from ..cleanup_control import CleanupFailures
+from . import lifecycle_lock, util
 from .bottle_cleanup_plan import FirecrackerBottleCleanupPlan
 
 
@@ -38,7 +38,7 @@ def _run_root() -> Path:
     return util.cache_dir() / "run"
 
 
-def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
+def _run_dir_of(args: Sequence[str], run_root: Path) -> Path | None:
     """The bottle run dir a firecracker cmdline belongs to, or None.
 
     A bottle VM is launched with `--config-file <run_root>/<slug>/config.json`,
@@ -46,15 +46,35 @@ def _run_dir_of(cmd: str, run_root: Path) -> Path | None:
     the run root. Anything else (a builder VM, the infra VM elsewhere) is
     not ours to reap here.
     """
-    toks = cmd.split()
-    for i, tok in enumerate(toks):
-        if tok == "--config-file" and i + 1 < len(toks):
-            parent = Path(toks[i + 1]).parent
+    for i, arg in enumerate(args):
+        if arg == "--config-file" and i + 1 < len(args):
+            parent = Path(args[i + 1]).parent
             if parent.parent == run_root:
                 return parent
     return None
 
 
+def _decode_cmdline(raw: bytes) -> tuple[str, ...]:
+    """Decode Linux's NUL-delimited argv without losing embedded spaces."""
+    return tuple(
+        value.decode(errors="surrogateescape")
+        for value in raw.split(b"\0") if value
+    )
+
+
+def _process_args(pid: int) -> tuple[str, ...] | None:
+    """Read one process's lossless argv, or None when it exited meanwhile."""
+    try:
+        raw = Path(f"/proc/{pid}/cmdline").read_bytes()
+    except FileNotFoundError:
+        return None
+    except OSError as exc:
+        raise EnumerationError(
+            f"could not inspect Firecracker pid {pid}: {exc}"
+        ) from exc
+    return _decode_cmdline(raw)
+
+
 def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     """Inspect running firecracker VMs under ``run_root``.
 
@@ -65,7 +85,7 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     """
     try:
         result = subprocess.run(
-            ["pgrep", "-a", "firecracker"],
+            ["pgrep", "firecracker"],
             capture_output=True, text=True, check=False,
         )
     except OSError as exc:
@@ -83,14 +103,14 @@ def _scan_processes(run_root: Path) -> tuple[set[str], list[int]]:
     live: set[str] = set()
     orphan_pids: list[int] = []
     for line in result.stdout.splitlines():
-        parts = line.split(None, 1)
-        if len(parts) != 2:
-            continue
         try:
-            pid = int(parts[0])
+            pid = int(line.strip())
         except ValueError:
             continue
-        run_dir = _run_dir_of(parts[1], run_root)
+        args = _process_args(pid)
+        if args is None:
+            continue
+        run_dir = _run_dir_of(args, run_root)
         if run_dir is None:
             continue
         if run_dir.is_dir():
@@ -126,12 +146,44 @@ def prepare_cleanup() -> FirecrackerBottleCleanupPlan:
 
 
 def cleanup(plan: FirecrackerBottleCleanupPlan) -> None:
-    for pid in plan.vm_pids:
-        info(f"kill firecracker VM pid {pid}")
-        try:
-            os.kill(pid, signal.SIGTERM)
-        except ProcessLookupError:
-            pass
-    for path in plan.run_dirs:
+    """Revalidate the preview under the launch lock, then remove its survivors."""
+    with lifecycle_lock.hold():
+        fresh = prepare_cleanup()
+        approved_pids = set(plan.vm_pids).intersection(fresh.vm_pids)
+        approved_dirs = set(plan.run_dirs).intersection(fresh.run_dirs)
+        failures = CleanupFailures()
+        for pid in sorted(approved_pids):
+            _terminate_orphan(pid, _run_root())
+        for path in sorted(approved_dirs):
             info(f"rm -rf {path}")
-        shutil.rmtree(path, ignore_errors=True)
+            failures.remove_tree(Path(path), f"removing Firecracker run dir {path}")
+        failures.raise_if_any()
+
+
+def _terminate_orphan(pid: int, run_root: Path) -> None:
+    """Signal exactly the process identity that still owns an orphan config."""
+    try:
+        pidfd = os.pidfd_open(pid)
+    except ProcessLookupError:
+        return
+    except OSError as exc:
+        raise EnumerationError(
+            f"could not pin Firecracker pid {pid} for cleanup: {exc}"
+        ) from exc
+    try:
+        try:
+            raw = Path(f"/proc/{pid}/cmdline").read_bytes()
+        except FileNotFoundError:
+            return
+        except OSError as exc:
+            raise EnumerationError(
+                f"could not revalidate Firecracker pid {pid}: {exc}"
+            ) from exc
+        args = _decode_cmdline(raw)
+        run_dir = _run_dir_of(args, run_root)
+        if run_dir is None or run_dir.is_dir():
+            return
+        info(f"kill firecracker VM pid {pid}")
+        signal.pidfd_send_signal(pidfd, signal.SIGTERM)
+    finally:
+        os.close(pidfd)

bot_bottle/backend/firecracker/infra_artifact.py

@@ -41,6 +41,8 @@ from ... import resources
 from ...log import die, info
 from . import util
 
+ARTIFACT_HTTP_TIMEOUT_SECONDS = 30.0
+
 # Bump if the on-disk artifact *format* changes (compression, layout) so a new
 # scheme can't collide with a cached/published artifact of the old one.
 _ARTIFACT_FORMAT = "1"
@@ -164,7 +166,9 @@ def _download(url: str, dest: Path) -> None:
     """Stream `url` to `dest` (atomic via a `.part` sibling)."""
     tmp = dest.with_suffix(dest.suffix + ".part")
     try:
-        with urllib.request.urlopen(_open(url)) as resp, open(tmp, "wb") as out:
+        with urllib.request.urlopen(
+            _open(url), timeout=ARTIFACT_HTTP_TIMEOUT_SECONDS,
+        ) as resp, open(tmp, "wb") as out:
             shutil.copyfileobj(resp, out, _CHUNK)
     except urllib.error.HTTPError as e:
         tmp.unlink(missing_ok=True)

bot_bottle/backend/firecracker/launch.py

@@ -46,7 +46,7 @@ from ...log import die, info, warn
 from ...supervisor.types import SUPERVISE_PORT
 from ..docker.egress import EGRESS_PORT
 from ..util import AGENT_CA_BUNDLE, AGENT_CA_PATH
-from . import firecracker_vm, image_builder, isolation_probe, netpool, util
+from . import firecracker_vm, image_builder, isolation_probe, lifecycle_lock, netpool, util
 from .bottle import FirecrackerBottle
 from .bottle_plan import FirecrackerBottlePlan
 from ...orchestrator.store.config_store import resolve_teardown_timeout
@@ -164,11 +164,15 @@ def launch(
         )
 
         # Step 6: build the per-bottle rootfs + SSH key, then boot.
+        # Cleanup takes the same lock while refreshing its process snapshot.
+        # Hold it until the VMM exists so a newly-created run dir can never be
+        # mistaken for an orphan in the build-before-boot window.
+        with lifecycle_lock.hold():
             run_dir = util.cache_dir() / "run" / plan.slug
             run_dir.mkdir(parents=True, exist_ok=True)
             # Remove the run dir on teardown so the per-bottle rootfs.ext4 (~1G)
-        # doesn't leak. Registered before vm.terminate below so it runs *after*
-        # it (ExitStack is LIFO): the VM is gone before we rm its rootfs.
+            # doesn't leak. Registered before vm.terminate below so it runs
+            # *after* it (ExitStack is LIFO).
             stack.callback(lambda: shutil.rmtree(run_dir, ignore_errors=True))
             rootfs = run_dir / "rootfs.ext4"
             util.build_rootfs_ext4(agent_base, rootfs)

bot_bottle/backend/firecracker/lifecycle_lock.py

@@ -0,0 +1,30 @@
+"""Serialize Firecracker run-directory creation with orphan cleanup."""
+
+from __future__ import annotations
+
+import fcntl
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Generator
+
+from . import util
+
+
+def _lock_path() -> Path:
+    return util.cache_dir() / "run.lifecycle.lock"
+
+
+@contextmanager
+def hold() -> Generator[None]:
+    """Exclude cleanup while a launch directory lacks a visible VMM."""
+    path = _lock_path()
+    path.parent.mkdir(parents=True, exist_ok=True)
+    with path.open("a", encoding="utf-8") as handle:
+        fcntl.flock(handle, fcntl.LOCK_EX)
+        try:
+            yield
+        finally:
+            fcntl.flock(handle, fcntl.LOCK_UN)
+
+
+__all__ = ["hold"]

bot_bottle/backend/firecracker/publish_infra.py

@@ -34,6 +34,7 @@ from pathlib import Path
 from . import infra_artifact, infra_vm, util
 
 _CHUNK = 1 << 20
+_REGISTRY_HTTP_TIMEOUT_SECONDS = 30.0
 
 _GZ_NAME = "rootfs.ext4.gz"
 _SHA_NAME = "rootfs.ext4.gz.sha256"
@@ -91,7 +92,9 @@ def _put(url: str, body: "bytes | Path", token: str) -> None:
         req.add_header("Authorization", f"token {token}")
     req.add_header("Content-Type", "application/octet-stream")
     try:
-        with urllib.request.urlopen(req) as resp:
+        with urllib.request.urlopen(
+            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ) as resp:
             print(f"  uploaded {url} (HTTP {resp.status})")
     except urllib.error.HTTPError as e:
         if e.code == 409:
@@ -112,7 +115,9 @@ def _delete(url: str, token: str) -> None:
     if token:
         req.add_header("Authorization", f"token {token}")
     try:
-        with urllib.request.urlopen(req):
+        with urllib.request.urlopen(
+            req, timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ):
             pass
     except urllib.error.HTTPError as e:
         if e.code != 404:
@@ -151,7 +156,10 @@ def _try_download_published(role: str, role_dir: Path) -> str | None:
     version = _role_version(role)
     sha_url = infra_artifact.artifact_url(version, _SHA_NAME, role=role)
     try:
-        with urllib.request.urlopen(infra_artifact._open(sha_url)):
+        with urllib.request.urlopen(
+            infra_artifact._open(sha_url),
+            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ):
             pass
     except urllib.error.HTTPError as e:
         if e.code == 404:
@@ -195,7 +203,10 @@ def _publish_bundle(role: str, role_dir: Path, token: str) -> str:
     # present, a re-publish is a no-op. Otherwise clear any partial upload left
     # by an interrupted prior attempt and upload the complete set.
     try:
-        with urllib.request.urlopen(infra_artifact._open(sha_url)) as resp:
+        with urllib.request.urlopen(
+            infra_artifact._open(sha_url),
+            timeout=_REGISTRY_HTTP_TIMEOUT_SECONDS,
+        ) as resp:
             remote_sha = resp.read().decode("utf-8").split()[0].strip().lower()
     except urllib.error.HTTPError as e:
         if e.code != 404:

bot_bottle/backend/macos_container/bottle_cleanup_plan.py

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

bot_bottle/backend/macos_container/cleanup.py

@@ -4,7 +4,9 @@ from __future__ import annotations
 
 import subprocess
 
-from ...log import info, warn
+from .. import EnumerationError
+from ..cleanup_control import CleanupFailures
+from ...log import info
 from . import util as container_mod
 from .bottle_cleanup_plan import MacosContainerBottleCleanupPlan
 
@@ -19,8 +21,8 @@ def _list_prefixed_containers() -> list[str]:
         check=False,
     )
     if result.returncode != 0:
-        warn(f"container list failed: {result.stderr.strip()}")
-        return []
+        detail = result.stderr.strip() or f"exit {result.returncode}"
+        raise EnumerationError(f"container list failed: {detail}")
     return sorted(
         name for name in (line.strip() for line in result.stdout.splitlines())
         if name.startswith(_PREFIX)
@@ -35,7 +37,8 @@ def _list_prefixed_networks() -> list[str]:
         check=False,
     )
     if result.returncode != 0:
-        return []
+        detail = result.stderr.strip() or f"exit {result.returncode}"
+        raise EnumerationError(f"container network list failed: {detail}")
     return sorted(
         name for name in (line.strip() for line in result.stdout.splitlines())
         if name.startswith(_PREFIX)
@@ -51,19 +54,17 @@ def prepare_cleanup() -> MacosContainerBottleCleanupPlan:
 
 
 def cleanup(plan: MacosContainerBottleCleanupPlan) -> None:
+    failures = CleanupFailures()
     for name in plan.containers:
         info(f"container delete --force {name}")
-        subprocess.run(
+        failures.run(
             ["container", "delete", "--force", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"deleting container {name}",
         )
     for name in plan.networks:
         info(f"container network delete {name}")
-        subprocess.run(
+        failures.run(
             ["container", "network", "delete", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-            check=False,
+            f"deleting network {name}",
         )
+    failures.raise_if_any()

bot_bottle/cli/commands/cleanup.py

@@ -22,6 +22,7 @@ from __future__ import annotations
 import sys
 
 from ...backend import get_bottle_backend, has_backend, known_backend_names
+from ...backend.cleanup_control import CleanupError
 from ...log import info
 from ...util import read_tty_line
 
@@ -52,10 +53,20 @@ def cmd_cleanup(_argv: list[str]) -> int:
         info("cleanup: skipped")
         return 0
 
-    for name, backend, plan in prepared:
-        if plan.empty:
+    # Confirmation authorizes a fresh authoritative snapshot, not blind use of
+    # identities that may have changed while the operator reviewed the preview.
+    failures: list[str] = []
+    for name, backend, displayed in prepared:
+        current = backend.prepare_cleanup()
+        approved = displayed.intersect(current)
+        if approved.empty:
             continue
-        backend.cleanup(plan)
+        try:
+            backend.cleanup(approved)
+        except CleanupError as exc:
+            failures.append(f"{name}: {exc}")
+    if failures:
+        raise CleanupError("cleanup incomplete: " + "; ".join(failures))
     info("cleanup: done")
     return 0
 

bot_bottle/gateway/bootstrap.py

@@ -136,10 +136,18 @@ def _pump(name: str, stream: IO[bytes]) -> None:
     """Read lines from `stream`, prefix with `[name]`, write to
     stdout. Runs in its own thread per child; daemon=True so a
     blocked read doesn't keep the process alive after main exits."""
+    try:
         for raw in iter(stream.readline, b""):
             line = raw.decode("utf-8", errors="replace").rstrip("\n")
             sys.stdout.write(f"[{name}] {line}\n")
             sys.stdout.flush()
+    except (OSError, ValueError) as exc:
+        # The manager closes a dead child's pipe after wait() and before a
+        # restart. A pump can be between readline calls at that exact moment;
+        # closed-stream errors are normal completion, not uncaught thread
+        # failures. Preserve genuinely unexpected I/O diagnostics.
+        if not stream.closed:
+            _log(f"{name} output pump stopped: {type(exc).__name__}: {exc}")
 
 
 def _spawn(spec: _DaemonSpec) -> subprocess.Popen[bytes]:

bot_bottle/gateway/bounded_http.py

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

bot_bottle/gateway/git_gate/http_backend.py

@@ -22,11 +22,16 @@ import os
 import subprocess
 import sys
 import typing
-from http.server import BaseHTTPRequestHandler, ThreadingHTTPServer
+from http.server import BaseHTTPRequestHandler
 from pathlib import Path
 from urllib.parse import urlsplit
 
 from bot_bottle.constants import GIT_GATE_TIMEOUT_SECS, IDENTITY_HEADER
+from bot_bottle.gateway.bounded_http import (
+    BodyReadError,
+    BoundedThreadingHTTPServer,
+    read_declared_body,
+)
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 
 
@@ -77,6 +82,8 @@ def resolve_sandbox_root(
 
 # Bound memory use while still allowing ordinary git push packfiles.
 MAX_BODY_BYTES = 100 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 30.0
+MAX_REQUEST_WORKERS = 16
 
 
 class GitHttpHandler(BaseHTTPRequestHandler):
@@ -184,19 +191,19 @@ class GitHttpHandler(BaseHTTPRequestHandler):
             value = self.headers.get(header)
             if value:
                 env[variable] = value
-        raw_length = self.headers.get("content-length", "0") or "0"
         try:
-            length = int(raw_length)
-        except ValueError:
-            self.send_error(400, "Bad Content-Length")
+            body = read_declared_body(
+                self.rfile,
+                self.connection,
+                self.headers.get("content-length"),
+                maximum=MAX_BODY_BYTES,
+                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                require_length=False,
+            )
+        except BodyReadError as exc:
+            self.send_error(exc.status, exc.message)
             return
-        if length < 0:
-            self.send_error(400, "Negative Content-Length")
-            return
-        if length > MAX_BODY_BYTES:
-            self.send_error(413, "Request body too large")
-            return
-        body = self.rfile.read(length) if length else b""
+        try:
             proc = subprocess.run(
                 ["git", "http-backend"],
                 input=body,
@@ -205,6 +212,10 @@ class GitHttpHandler(BaseHTTPRequestHandler):
                 check=False,
                 timeout=GIT_GATE_TIMEOUT_SECS,
             )
+        except (OSError, subprocess.SubprocessError) as exc:
+            self.log_message("git http-backend unavailable: %s", exc)
+            self.send_error(503, "git backend unavailable")
+            return
         self._write_cgi_response(proc.stdout)
 
     def _repo_dir(self, sandbox_root: Path, path: str) -> Path | None:
@@ -273,7 +284,9 @@ def main() -> int:
             "(no single-tenant flat-root fallback)\n"
         )
         return 1
-    server = ThreadingHTTPServer(("0.0.0.0", port), GitHttpHandler)
+    server = BoundedThreadingHTTPServer(
+        ("0.0.0.0", port), GitHttpHandler, max_workers=MAX_REQUEST_WORKERS,
+    )
     # Resolve each request's sandbox namespace by source IP against the
     # orchestrator control plane.
     server.policy_resolver = PolicyResolver(orch_url)  # type: ignore[attr-defined]

bot_bottle/gateway/supervisor/mcp_dispatch.py

@@ -6,9 +6,8 @@ import json
 from dataclasses import dataclass
 from typing import Callable, Protocol
 
-from bot_bottle.gateway.egress.context import resolve_client_context
-from bot_bottle.gateway.egress.schema import route_to_yaml_dict
-from bot_bottle.gateway.policy_resolver import PolicyResolver
+from bot_bottle.gateway.egress.schema import load_config, route_to_yaml_dict
+from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 from bot_bottle.supervisor import types as _sv
 
 
@@ -24,6 +23,10 @@ class MethodNotFoundError(Exception):
     """Raised when a JSON-RPC method has no MCP handler."""
 
 
+class RouteResolutionError(Exception):
+    """The caller's live route table could not be resolved authoritatively."""
+
+
 Handler = Callable[[dict[str, object]], object]
 
 
@@ -60,10 +63,19 @@ def resolved_routes_payload(
     source_ip: str,
     identity_token: str,
 ) -> dict[str, object]:
-    """Render the calling bottle's routes, failing closed to an empty list."""
-    config, _slug, _tokens = resolve_client_context(
-        resolver, source_ip, identity_token,
+    """Render an authoritatively resolved route table for the calling bottle."""
+    try:
+        policy, bottle_id, _tokens = resolver.resolve_policy_and_bottle_id(
+            source_ip, identity_token,
         )
+    except PolicyResolveError as exc:
+        raise RouteResolutionError("orchestrator unavailable") from exc
+    if not bottle_id:
+        raise RouteResolutionError("request source is not attributed to a bottle")
+    try:
+        config = load_config(policy or "")
+    except ValueError as exc:
+        raise RouteResolutionError("resolved policy is invalid") from exc
     body = json.dumps(
         {"routes": [route_to_yaml_dict(route) for route in config.routes]},
         indent=2,
@@ -74,6 +86,7 @@ def resolved_routes_payload(
 __all__ = [
     "Handlers",
     "MethodNotFoundError",
+    "RouteResolutionError",
     "dispatch",
     "resolved_routes_payload",
 ]

bot_bottle/gateway/supervisor/server.py

@@ -51,19 +51,24 @@ from __future__ import annotations
 import http.server
 import json
 import os
-import socketserver
 import sys
 import time
 import typing
 from dataclasses import dataclass
 
 from bot_bottle.constants import IDENTITY_HEADER
+from bot_bottle.gateway.bounded_http import (
+    BodyReadError,
+    BoundedThreadingHTTPServer,
+    read_declared_body,
+)
 from bot_bottle.gateway.egress.schema import load_config
 from bot_bottle.gateway.egress.types import LOG_OFF
 from bot_bottle.gateway.policy_resolver import PolicyResolveError, PolicyResolver
 from bot_bottle.gateway.supervisor.mcp_dispatch import (
     Handlers as DispatchHandlers,
     MethodNotFoundError,
+    RouteResolutionError,
     dispatch,
     resolved_routes_payload,
 )
@@ -570,6 +575,8 @@ def format_unknown_proposal_text(proposal_id: str) -> str:
 # Max request body the server accepts. 1 MB is well above any realistic
 # routes.yaml proposal.
 MAX_BODY_BYTES = 1 * 1024 * 1024
+REQUEST_BODY_TIMEOUT_SECONDS = 10.0
+MAX_REQUEST_WORKERS = 32
 
 
 class MCPHandler(http.server.BaseHTTPRequestHandler):
@@ -592,19 +599,18 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
         self._write_text(405, "use POST for MCP requests\n")
 
     def do_POST(self) -> None:
-        length_header = self.headers.get("Content-Length")
-        if length_header is None:
-            self._write_text(411, "Content-Length required\n")
-            return
         try:
-            length = int(length_header)
-        except ValueError:
-            self._write_text(400, "invalid Content-Length\n")
+            body = read_declared_body(
+                self.rfile,
+                self.connection,
+                self.headers.get("Content-Length"),
+                maximum=MAX_BODY_BYTES,
+                timeout_seconds=REQUEST_BODY_TIMEOUT_SECONDS,
+                require_length=True,
+            )
+        except BodyReadError as exc:
+            self._write_text(exc.status, exc.message + "\n")
             return
-        if length < 0 or length > MAX_BODY_BYTES:
-            self._write_text(413, "request body too large\n")
-            return
-        body = self.rfile.read(length)
 
         try:
             req = parse_jsonrpc(body)
@@ -660,20 +666,25 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
                 identity_token=self._identity_token(),
             )
 
-        return dispatch(
-            req,
-            DispatchHandlers(
-                initialize=handle_initialize,
-                tools_list=handle_tools_list,
-                list_routes=lambda _params: resolved_routes_payload(
+        def list_routes(_params: dict[str, object]) -> object:
+            try:
+                return resolved_routes_payload(
                     self._resolver_or_fail(),
                     self.client_address[0],
                     self._identity_token(),
-                ),
+                )
+            except RouteResolutionError as exc:
+                raise _RpcInternalError(
+                    f"could not resolve live egress routes: {exc}"
+                ) from exc
+
+        return dispatch(req, DispatchHandlers(
+            initialize=handle_initialize,
+            tools_list=handle_tools_list,
+            list_routes=list_routes,
             check_proposal=check,
             propose=propose,
-            ),
-        )
+        ))
 
     def _identity_token(self) -> str:
         """The agent's per-bottle identity token from the request header (the
@@ -711,7 +722,7 @@ class MCPHandler(http.server.BaseHTTPRequestHandler):
             self.wfile.write(encoded)
 
 
-class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
+class MCPServer(BoundedThreadingHTTPServer):
     allow_reuse_address = True
     daemon_threads = True
     config: ServerConfig = ServerConfig()
@@ -720,6 +731,9 @@ class MCPServer(socketserver.ThreadingMixIn, http.server.HTTPServer):
     # closed per request (see `_resolver_or_fail`).
     policy_resolver: "PolicyResolver | None" = None
 
+    def __init__(self, *args, **kwargs):  # type: ignore[no-untyped-def]
+        super().__init__(*args, max_workers=MAX_REQUEST_WORKERS, **kwargs)
+
 
 # --- Entry point -----------------------------------------------------------
 

bot_bottle/orchestrator/service.py

@@ -366,7 +366,7 @@ class OrchestratorCore:
         value with *env_var_secret*, and restores ``_tokens[bottle_id]``.
         Returns True on success, False when no stored secrets exist for this
         bottle or decryption fails (wrong key / corrupt data)."""
-        from .store.secret_store import decrypt_value, encrypt_value, is_legacy_blob
+        from .store.secret_store import decrypt_value
         encrypted = self.registry.get_agent_secrets(bottle_id)
         if not encrypted:
             return False
@@ -377,12 +377,6 @@ class OrchestratorCore:
         except ValueError:
             return False
         self._tokens[bottle_id] = decrypted
-        if any(is_legacy_blob(value) for value in encrypted.values()):
-            migrated = {
-                key: encrypt_value(env_var_secret, value)
-                for key, value in decrypted.items()
-            }
-            self.registry.store_agent_secrets(bottle_id, migrated)
         return True
 
     # --- consolidated gateway ----------------------------------------------

bot_bottle/orchestrator/store/registry_store.py

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

bot_bottle/orchestrator/store/secret_store.py

@@ -18,9 +18,9 @@ value is encrypted independently. New output blobs are:
 
 ``version || nonce (16 bytes) || ciphertext || tag (32 bytes)``
 
-encoded as URL-safe base64 (no padding). The version marker lets the reader
-accept legacy ``nonce || ciphertext`` rows long enough to rewrite them in the
-authenticated format after a successful reprovision.
+encoded as URL-safe base64 (no padding). Unversioned legacy ciphertext is
+rejected; the registry migration clears those rows rather than allowing blob
+contents to select an unauthenticated decoder.
 
     keystream_block_i = HMAC-SHA256(key, nonce || i.to_bytes(4, "big"))
     ciphertext_i      = plaintext_i XOR keystream_block_i[:len(plaintext_i)]
@@ -84,44 +84,18 @@ def encrypt_value(secret_b64: str, plaintext: str) -> str:
     return base64.urlsafe_b64encode(authenticated + tag).rstrip(b"=").decode()
 
 
-def is_legacy_blob(blob_b64: str) -> bool:
-    """Whether *blob_b64* uses the pre-authentication storage format."""
-    try:
-        return not _b64dec(blob_b64).startswith(_VERSION)
-    except (ValueError, TypeError):
-        return False
-
-
-def _decrypt_legacy(key: bytes, blob: bytes) -> str:
-    """Read the original ``nonce || ciphertext`` format for migration only."""
-    if len(blob) < _NONCE_BYTES:
-        raise ValueError("ciphertext blob too short")
-    nonce, ciphertext = blob[:_NONCE_BYTES], blob[_NONCE_BYTES:]
-    pt = bytearray()
-    # The legacy format used the byte offset as the PRF counter.
-    for i in range(0, len(ciphertext), _BLOCK):
-        chunk = ciphertext[i : i + _BLOCK]
-        ks = _keystream(key, nonce, i)[: len(chunk)]
-        pt.extend(c ^ k for c, k in zip(chunk, ks))
-    try:
-        return bytes(pt).decode()
-    except UnicodeDecodeError as exc:
-        raise ValueError(f"decryption produced non-UTF-8 output: {exc}") from exc
-
-
 def decrypt_value(secret_b64: str, blob_b64: str) -> str:
     """Decrypt a blob produced by :func:`encrypt_value`.
 
     Returns the original plaintext string. Raises ``ValueError`` for malformed
-    input, authentication failure, or a key mismatch. Legacy unauthenticated
-    rows remain readable so callers can migrate them immediately."""
+    input, authentication failure, or a key mismatch."""
     key = _b64dec(secret_b64)
     try:
         blob = _b64dec(blob_b64)
     except (ValueError, TypeError) as exc:
         raise ValueError(f"invalid ciphertext blob: {exc}") from exc
     if not blob.startswith(_VERSION):
-        return _decrypt_legacy(key, blob)
+        raise ValueError("unsupported ciphertext format")
     minimum = len(_VERSION) + _NONCE_BYTES + _TAG_BYTES
     if len(blob) < minimum:
         raise ValueError("ciphertext blob too short")
@@ -152,5 +126,4 @@ __all__ = [
     "new_env_var_secret",
     "encrypt_value",
     "decrypt_value",
-    "is_legacy_blob",
 ]

tests/unit/test_cli_cleanup_cross_backend.py

@@ -17,6 +17,7 @@ from bot_bottle.cli.commands import cleanup as cmd
 def _make_backend(empty: bool = True):
     backend = MagicMock()
     plan = MagicMock(empty=empty)
+    plan.intersect.return_value = plan
     backend.prepare_cleanup.return_value = plan
     backend.cleanup = MagicMock()
     return backend, plan
@@ -41,8 +42,8 @@ class TestCmdCleanup(unittest.TestCase):
         ):
             self.assertEqual(0, cmd.cmd_cleanup([]))
 
-        docker.prepare_cleanup.assert_called_once()
-        fc.prepare_cleanup.assert_called_once()
+        self.assertEqual(2, docker.prepare_cleanup.call_count)
+        self.assertEqual(2, fc.prepare_cleanup.call_count)
         docker.cleanup.assert_called_once_with(docker_plan)
         fc.cleanup.assert_called_once_with(fc_plan)
 
@@ -68,7 +69,7 @@ class TestCmdCleanup(unittest.TestCase):
         ):
             self.assertEqual(0, cmd.cmd_cleanup([]))
 
-        docker.prepare_cleanup.assert_called_once()
+        self.assertEqual(2, docker.prepare_cleanup.call_count)
         docker.cleanup.assert_called_once_with(docker_plan)
         macos.prepare_cleanup.assert_not_called()
 
@@ -135,6 +136,28 @@ class TestCmdCleanup(unittest.TestCase):
         docker.cleanup.assert_called_once_with(docker_plan)
         fc.cleanup.assert_not_called()
 
+    def test_executes_only_displayed_resources_still_current(self):
+        backend = MagicMock()
+        preview = MagicMock(empty=False)
+        refreshed = MagicMock(empty=False)
+        approved = MagicMock(empty=False)
+        preview.intersect.return_value = approved
+        backend.prepare_cleanup.side_effect = [preview, refreshed]
+
+        with patch.object(
+            cmd, "known_backend_names", return_value=("firecracker",),
+        ), patch.object(
+            cmd, "get_bottle_backend", return_value=backend,
+        ), patch.object(
+            cmd, "has_backend", return_value=True,
+        ), patch.object(
+            cmd, "_prompt_yes", return_value=True,
+        ):
+            self.assertEqual(0, cmd.cmd_cleanup([]))
+
+        preview.intersect.assert_called_once_with(refreshed)
+        backend.cleanup.assert_called_once_with(approved)
+
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_docker_cleanup.py

@@ -14,9 +14,12 @@ from __future__ import annotations
 import tempfile
 import unittest
 from pathlib import Path
+from unittest.mock import patch
 
 from tests.unit import use_bottle_root
 from bot_bottle import bottle_state
+from bot_bottle.backend import EnumerationError
+from bot_bottle.backend.docker import cleanup
 from bot_bottle.backend.docker.cleanup import _list_orphan_state_dirs
 
 
@@ -116,5 +119,37 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         )
 
 
+class TestAuthoritativeDiscovery(unittest.TestCase):
+    def test_prepare_requires_authoritative_compose_projects(self):
+        with patch.object(cleanup.docker_mod, "require_docker"), \
+             patch.object(
+                 cleanup, "list_compose_projects",
+                 side_effect=EnumerationError("compose unavailable"),
+             ) as projects, self.assertRaisesRegex(
+                 EnumerationError, "compose unavailable",
+             ):
+            cleanup.prepare_cleanup()
+        projects.assert_called_once_with(
+            warn_on_error=False,
+            raise_on_error=True,
+        )
+
+    def test_container_query_failure_raises(self):
+        failed = cleanup.subprocess.CompletedProcess(
+            [], 1, stdout="", stderr="daemon unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=failed), \
+             self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
+            cleanup._list_prefixed_containers()
+
+    def test_network_query_failure_raises(self):
+        failed = cleanup.subprocess.CompletedProcess(
+            [], 1, stdout="", stderr="daemon unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=failed), \
+             self.assertRaisesRegex(EnumerationError, "daemon unavailable"):
+            cleanup._list_prefixed_networks()
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_firecracker_cleanup.py

@@ -32,15 +32,19 @@ class TestProcessScan(unittest.TestCase):
         self.assertEqual(
             Path("/cache/run/dev-a"),
             fc_cleanup._run_dir_of(
-                f"firecracker --config-file {run_root}/dev-a/config.json", run_root
+                ("firecracker", "--config-file",
+                 f"{run_root}/dev-a/config.json"), run_root
             ),
         )
         # infra/builder VMs elsewhere, or nested paths, are not ours.
         self.assertIsNone(
-            fc_cleanup._run_dir_of("firecracker --config-file /elsewhere/config.json", run_root)
+            fc_cleanup._run_dir_of(
+                ("firecracker", "--config-file", "/elsewhere/config.json"),
+                run_root,
+            )
         )
         self.assertIsNone(
-            fc_cleanup._run_dir_of("firecracker --no-config", run_root)
+            fc_cleanup._run_dir_of(("firecracker", "--no-config"), run_root)
         )
 
     def test_scan_splits_live_dirs_from_orphan_pids(self):
@@ -48,17 +52,40 @@ class TestProcessScan(unittest.TestCase):
             run_root = Path(tmp)
             (run_root / "live-a").mkdir()  # dir present -> live VM, protected
             # "gone-b" dir intentionally absent -> lingering VMM, orphan pid
-            out = (
-                f"111 firecracker --config-file {run_root}/live-a/config.json\n"
-                f"222 firecracker --config-file {run_root}/gone-b/config.json\n"
-                "333 firecracker --config-file /elsewhere/config.json\n"
-                "notanint firecracker --config-file x\n"
-            )
-            with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)):
+            args = {
+                111: ("firecracker", "--config-file",
+                      f"{run_root}/live-a/config.json"),
+                222: ("firecracker", "--config-file",
+                      f"{run_root}/gone-b/config.json"),
+                333: ("firecracker", "--config-file", "/elsewhere/config.json"),
+            }
+            with patch.object(
+                fc_cleanup.subprocess, "run",
+                return_value=_proc("111\n222\n333\nnotanint\n"),
+            ), patch.object(
+                fc_cleanup, "_process_args", side_effect=args.get,
+            ):
                 live, orphan_pids = fc_cleanup._scan_processes(run_root)
         self.assertEqual({str(run_root / "live-a")}, live)
         self.assertEqual([222], orphan_pids)
 
+    def test_scan_preserves_spaces_in_config_path(self):
+        with tempfile.TemporaryDirectory(prefix="fc cache ") as tmp:
+            run_root = Path(tmp)
+            live = run_root / "live bottle"
+            live.mkdir()
+            with patch.object(
+                fc_cleanup.subprocess, "run", return_value=_proc("111\n"),
+            ), patch.object(
+                fc_cleanup, "_process_args",
+                return_value=("firecracker", "--config-file",
+                              str(live / "config.json")),
+            ):
+                self.assertEqual(
+                    ({str(live)}, []),
+                    fc_cleanup._scan_processes(run_root),
+                )
+
     def test_scan_empty_when_pgrep_finds_no_processes(self):
         with patch.object(fc_cleanup.subprocess, "run", return_value=_proc(returncode=1)):
             self.assertEqual((set(), []), fc_cleanup._scan_processes(Path("/x")))
@@ -117,9 +144,14 @@ class TestProcessScan(unittest.TestCase):
             run_root = Path(tmp)
             (run_root / "live-a").mkdir()
             (run_root / "dead-b").mkdir()
-            out = f"111 firecracker --config-file {run_root}/live-a/config.json\n"
             with patch.object(fc_cleanup, "_run_root", return_value=run_root), \
-                 patch.object(fc_cleanup.subprocess, "run", return_value=_proc(out)):
+                 patch.object(fc_cleanup.subprocess, "run",
+                              return_value=_proc("111\n")), \
+                 patch.object(
+                     fc_cleanup, "_process_args",
+                     return_value=("firecracker", "--config-file",
+                                   str(run_root / "live-a/config.json")),
+                 ):
                 plan = fc_cleanup.prepare_cleanup()
         self.assertEqual((), plan.vm_pids)
         self.assertEqual((str(run_root / "dead-b"),), plan.run_dirs)
@@ -132,18 +164,46 @@ class TestCleanupRemoval(unittest.TestCase):
             vm_pids=(101,),
             run_dirs=("/run/dev-x",),
         )
-        with patch.object(fc_cleanup.os, "kill") as kill, \
-             patch.object(fc_cleanup.shutil, "rmtree") as rmtree, \
+        with patch.object(fc_cleanup, "prepare_cleanup", return_value=plan), \
+             patch.object(fc_cleanup, "_run_root", return_value=Path("/run")), \
+             patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
+             patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree, \
              patch.object(fc_cleanup, "info"):
             fc_cleanup.cleanup(plan)
-        kill.assert_called_once()
-        rmtree.assert_called_once_with("/run/dev-x", ignore_errors=True)
+        terminate.assert_called_once_with(101, Path("/run"))
+        rmtree.assert_called_once_with(Path("/run/dev-x"))
 
-    def test_cleanup_tolerates_dead_pid(self):
-        plan = FirecrackerBottleCleanupPlan(vm_pids=(999,))
-        with patch.object(fc_cleanup.os, "kill", side_effect=ProcessLookupError), \
-             patch.object(fc_cleanup, "info"):
-            fc_cleanup.cleanup(plan)  # must not raise
+    def test_cleanup_skips_resources_no_longer_in_refreshed_plan(self):
+        preview = FirecrackerBottleCleanupPlan(
+            vm_pids=(999,), run_dirs=("/run/reused",),
+        )
+        with patch.object(
+            fc_cleanup, "prepare_cleanup",
+            return_value=FirecrackerBottleCleanupPlan(),
+        ), patch.object(fc_cleanup, "_terminate_orphan") as terminate, \
+             patch("bot_bottle.backend.cleanup_control.shutil.rmtree") as rmtree:
+            fc_cleanup.cleanup(preview)
+        terminate.assert_not_called()
+        rmtree.assert_not_called()
+
+    def test_pidfd_prevents_pid_reuse_from_signalling_unrelated_process(self):
+        with patch.object(fc_cleanup.os, "pidfd_open", return_value=7), \
+             patch.object(
+                 fc_cleanup.Path, "read_bytes",
+                 return_value=b"/usr/bin/python\0worker.py\0",
+             ), patch.object(fc_cleanup.signal, "pidfd_send_signal") as send, \
+             patch.object(fc_cleanup.os, "close"):
+            fc_cleanup._terminate_orphan(101, Path("/run"))
+        send.assert_not_called()
+
+    def test_pidfd_signals_revalidated_orphan(self):
+        command = b"firecracker\0--config-file\0/run/gone/config.json\0"
+        with patch.object(fc_cleanup.os, "pidfd_open", return_value=7), \
+             patch.object(fc_cleanup.Path, "read_bytes", return_value=command), \
+             patch.object(fc_cleanup.signal, "pidfd_send_signal") as send, \
+             patch.object(fc_cleanup.os, "close"), patch.object(fc_cleanup, "info"):
+            fc_cleanup._terminate_orphan(101, Path("/run"))
+        send.assert_called_once_with(7, fc_cleanup.signal.SIGTERM)
 
 
 class TestCleanupPlan(unittest.TestCase):

tests/unit/test_gateway_bounded_http.py

@@ -0,0 +1,79 @@
+"""Unit tests for shared gateway stdlib HTTP resource boundaries."""
+# pylint: disable=protected-access
+
+from __future__ import annotations
+
+import io
+import socket
+import unittest
+
+from bot_bottle.gateway.bounded_http import (
+    BodyReadError,
+    BoundedThreadingHTTPServer,
+    read_declared_body,
+)
+
+
+class _Handler:
+    pass
+
+
+class _TimeoutStream:
+    def read(self, _size: int = -1, /) -> bytes:
+        raise TimeoutError
+
+
+class TestDeclaredBody(unittest.TestCase):
+    def setUp(self) -> None:
+        self.left, self.right = socket.socketpair()
+
+    def tearDown(self) -> None:
+        self.left.close()
+        self.right.close()
+
+    def test_rejects_incomplete_body(self) -> None:
+        with self.assertRaisesRegex(BodyReadError, "incomplete"):
+            read_declared_body(
+                io.BytesIO(b"short"),
+                self.left,
+                "10",
+                maximum=100,
+                timeout_seconds=1,
+                require_length=True,
+            )
+
+    def test_maps_read_timeout(self) -> None:
+        with self.assertRaises(BodyReadError) as caught:
+            read_declared_body(
+                _TimeoutStream(),
+                self.left,
+                "1",
+                maximum=100,
+                timeout_seconds=1,
+                require_length=True,
+            )
+        self.assertEqual(408, caught.exception.status)
+
+
+class TestBoundedServer(unittest.TestCase):
+    def test_saturated_server_rejects_without_spawning_thread(self) -> None:
+        client, peer = socket.socketpair()
+        with BoundedThreadingHTTPServer(
+            ("127.0.0.1", 0), _Handler, max_workers=1,  # type: ignore[arg-type]
+        ) as server:
+            with client, peer:
+                # Directly reserve the only slot to model an in-flight handler.
+                self.assertTrue(
+                    server._request_slots.acquire(  # pylint: disable=consider-using-with
+                        blocking=False,
+                    ),
+                )
+                try:
+                    server.process_request(client, ("127.0.0.1", 1))
+                    self.assertIn(b"503 Service Unavailable", peer.recv(1024))
+                finally:
+                    server._request_slots.release()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_gateway_init.py

@@ -23,6 +23,7 @@ from bot_bottle.gateway.bootstrap import (
     _DaemonManager,
     _argv_for_daemon,
     _env_for_daemon,
+    _pump,
     _selected_daemons,
 )
 from tests._bin import SLEEP
@@ -565,5 +566,26 @@ class TestMainEndToEnd(unittest.TestCase):
         self.assertIn("no daemons selected", out)
 
 
+class _FailingStream:
+    def __init__(self, *, closed: bool) -> None:
+        self.closed = closed
+
+    def readline(self) -> bytes:
+        raise ValueError("I/O operation on closed file")
+
+
+class TestOutputPump(unittest.TestCase):
+    def test_closed_stream_race_is_normal_completion(self) -> None:
+        with patch("bot_bottle.gateway.bootstrap._log") as log:
+            _pump("egress", _FailingStream(closed=True))  # type: ignore[arg-type]
+        log.assert_not_called()
+
+    def test_unexpected_io_failure_is_logged(self) -> None:
+        with patch("bot_bottle.gateway.bootstrap._log") as log:
+            _pump("egress", _FailingStream(closed=False))  # type: ignore[arg-type]
+        log.assert_called_once()
+        self.assertIn("output pump stopped", log.call_args.args[0])
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_git_http_backend.py

@@ -486,6 +486,19 @@ class TestMalformedStatusHeader(unittest.TestCase):
         )
         self.assertEqual(500, status)
 
+    def test_backend_timeout_returns_503(self):
+        with mock.patch(
+            "bot_bottle.gateway.git_gate.http_backend.subprocess.run",
+            side_effect=subprocess.TimeoutExpired(["git", "http-backend"], 1),
+        ):
+            req = urllib.request.Request(
+                f"http://127.0.0.1:{self._port}/repo.git/info/refs",
+                method="GET",
+            )
+            with self.assertRaises(urllib.error.HTTPError) as raised:
+                urllib.request.urlopen(req, timeout=3)
+        self.assertEqual(503, raised.exception.code)
+
 
 class TestContentLengthBounds(unittest.TestCase):
     """PRD 0041: malformed or oversized Content-Length is rejected before

tests/unit/test_infra_artifact.py

@@ -284,6 +284,18 @@ class TestEnsureArtifact(_CacheMixin):
             ia.ensure_artifact_gz(version, role=_ROLE)
             self.assertEqual(first_calls, len(net.calls))
 
+    def test_download_uses_network_deadline(self) -> None:
+        response = mock.MagicMock()
+        response.__enter__.return_value = io.BytesIO(b"payload")
+        with tempfile.TemporaryDirectory() as d, mock.patch.object(
+            ia.urllib.request, "urlopen", return_value=response,
+        ) as urlopen:
+            ia._download("https://registry/artifact", Path(d) / "artifact")
+        self.assertEqual(
+            ia.ARTIFACT_HTTP_TIMEOUT_SECONDS,
+            urlopen.call_args.kwargs["timeout"],
+        )
+
     def test_checksum_mismatch_fails_closed(self) -> None:
         version = "beefbeefbeefbeef"
         gz = _gz(b"payload")

tests/unit/test_macos_container_cleanup.py

@@ -6,6 +6,7 @@ import unittest
 from unittest.mock import patch
 
 from bot_bottle.backend import EnumerationError
+from bot_bottle.backend.cleanup_control import CleanupError
 from bot_bottle.backend.macos_container import cleanup, enumerate as enum_mod
 from bot_bottle.backend.macos_container.bottle_cleanup_plan import (
     MacosContainerBottleCleanupPlan,
@@ -31,7 +32,10 @@ class TestMacosContainerCleanup(unittest.TestCase):
             containers=("bot-bottle-a",),
             networks=("bot-bottle-net-a",),
         )
-        with patch.object(cleanup.subprocess, "run") as run:
+        completed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=0, stdout="", stderr="",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=completed) as run:
             cleanup.cleanup(plan)
         self.assertEqual(
             ["container", "delete", "--force", "bot-bottle-a"],
@@ -42,6 +46,34 @@ class TestMacosContainerCleanup(unittest.TestCase):
             run.call_args_list[1].args[0],
         )
 
+    def test_cleanup_attempts_all_resources_then_raises(self):
+        plan = MacosContainerBottleCleanupPlan(
+            containers=("bot-bottle-a",), networks=("bot-bottle-net-a",),
+        )
+        failed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=failed) as run, \
+             self.assertRaisesRegex(CleanupError, "unavailable"):
+            cleanup.cleanup(plan)
+        self.assertEqual(2, run.call_count)
+
+    def test_container_enumeration_failure_aborts(self):
+        completed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="service unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=completed), \
+                self.assertRaisesRegex(EnumerationError, "service unavailable"):
+            cleanup._list_prefixed_containers()
+
+    def test_network_enumeration_failure_aborts(self):
+        completed = cleanup.subprocess.CompletedProcess(
+            args=[], returncode=1, stdout="", stderr="service unavailable",
+        )
+        with patch.object(cleanup.subprocess, "run", return_value=completed), \
+                self.assertRaisesRegex(EnumerationError, "service unavailable"):
+            cleanup._list_prefixed_networks()
+
 
 class TestMacosContainerEnumerate(unittest.TestCase):
     """The backend launches bottles again (PRD 0070), so enumeration is real

tests/unit/test_orchestrator_registry_store.py

@@ -224,6 +224,17 @@ class TestAgentSecrets(unittest.TestCase):
         reopened = RegistryStore(self.db)
         self.assertEqual({"K": "v"}, reopened.get_agent_secrets("bottle-1"))
 
+    def test_v6_migration_clears_legacy_secret_rows(self) -> None:
+        self.store.store_agent_secrets("bottle-1", {"K": "legacy"})
+        with closing(sqlite3.connect(self.db)) as conn:
+            conn.execute(
+                "UPDATE schema_versions SET version = 5 "
+                "WHERE module = 'orchestrator_registry'"
+            )
+            conn.commit()
+        self.store.migrate()
+        self.assertEqual({}, self.store.get_agent_secrets("bottle-1"))
+
 
 class TestReapAbsent(unittest.TestCase):
     """`reap_absent` — the self-heal for rows whose bottle is gone.

tests/unit/test_orchestrator_secret_store.py

@@ -3,8 +3,6 @@
 from __future__ import annotations
 
 import base64
-import hashlib
-import hmac
 import unittest
 
 from bot_bottle.orchestrator.store.secret_store import (
@@ -83,16 +81,21 @@ class TestDecryptErrors(unittest.TestCase):
         with self.assertRaisesRegex(ValueError, "authentication failed"):
             decrypt_value(self.secret, tampered)
 
-    def test_reads_legacy_ciphertext_for_migration(self) -> None:
-        key = base64.urlsafe_b64decode(self.secret + "==")
-        nonce = b"0123456789abcdef"
-        plaintext = b"legacy-token"
-        stream = hmac.new(
-            key, nonce + (0).to_bytes(4, "big"), hashlib.sha256,
-        ).digest()
-        ciphertext = bytes(p ^ k for p, k in zip(plaintext, stream))
-        legacy = base64.urlsafe_b64encode(nonce + ciphertext).rstrip(b"=").decode()
-        self.assertEqual("legacy-token", decrypt_value(self.secret, legacy))
+    def test_rejects_legacy_ciphertext(self) -> None:
+        legacy = base64.urlsafe_b64encode(
+            b"0123456789abcdeflegacy-token",
+        ).rstrip(b"=").decode()
+        with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
+            decrypt_value(self.secret, legacy)
+
+    def test_rejects_authenticated_blob_with_changed_version(self) -> None:
+        raw = bytearray(base64.urlsafe_b64decode(
+            encrypt_value(self.secret, "secret-token") + "=="
+        ))
+        raw[0] ^= 1
+        downgraded = base64.urlsafe_b64encode(raw).rstrip(b"=").decode()
+        with self.assertRaisesRegex(ValueError, "unsupported ciphertext format"):
+            decrypt_value(self.secret, downgraded)
 
     def test_truncated_blob_raises_value_error(self) -> None:
         with self.assertRaises(ValueError):

tests/unit/test_publish_infra.py

@@ -49,6 +49,16 @@ class TestPut(unittest.TestCase):
         self.assertNotIsInstance(req.data, (bytes, bytearray))
         self.assertEqual(str(len(payload)), req.get_header("Content-length"))
 
+    def test_put_uses_network_deadline(self) -> None:
+        with mock.patch.object(
+            pub.urllib.request, "urlopen", return_value=_Resp(),
+        ) as urlopen:
+            pub._put("https://reg/pkg", b"payload", token="")
+        self.assertEqual(
+            pub._REGISTRY_HTTP_TIMEOUT_SECONDS,
+            urlopen.call_args.kwargs["timeout"],
+        )
+
     def test_small_bytes_body_still_works(self) -> None:
         captured: list[urllib.request.Request] = []
 

tests/unit/test_supervise_server.py

@@ -717,10 +717,9 @@ class TestResolvedRoutesPayload(unittest.TestCase):
         hosts = {r["host"] for r in data["routes"]}
         self.assertEqual({"api.anthropic.com", "www.google.com"}, hosts)
 
-    def test_orchestrator_error_fails_closed_to_empty(self) -> None:
-        # resolve_client_context swallows resolver errors → deny-all (empty),
-        # never another bottle's routes.
-        payload = resolved_routes_payload(
+    def test_orchestrator_error_is_not_reported_as_empty_policy(self) -> None:
+        with self.assertRaises(supervise_server.RouteResolutionError):
+            resolved_routes_payload(
                 typing.cast(
                     supervise_server.PolicyResolver,
                     _FakeSuperviseResolver(raises=True),
@@ -728,7 +727,16 @@ class TestResolvedRoutesPayload(unittest.TestCase):
                 _SRC,
                 _TOK,
             )
-        assert payload is not None
+
+    def test_authoritative_empty_policy_remains_successful(self) -> None:
+        payload = resolved_routes_payload(
+            typing.cast(
+                supervise_server.PolicyResolver,
+                _FakeSuperviseResolver(bottle_id="b1", policy="routes: []\n"),
+            ),
+            _SRC,
+            _TOK,
+        )
         data = json.loads(payload["content"][0]["text"])  # type: ignore[index]
         self.assertEqual([], data["routes"])