refactor(smolmachines): drop obsolete docker-sidecar helpers

bot_bottle/backend/__init__.py

@@ -36,10 +36,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager, contextmanager
+from contextlib import AbstractContextManager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generator, Generic, Sequence, TypeVar
+from typing import Any, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -78,9 +78,6 @@ class BottleSpec:
     # True when launched via --headless (no TTY, no interactive prompts).
     # The git-gate host-key preflight uses this to error rather than prompt.
     headless: bool = False
-    # Image startup policy. "fresh" preserves the normal build path;
-    # "cached" reuses the current local image/artifact without rebuilding.
-    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -435,25 +432,8 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
         prompt file, Dockerfile path, and guest home all live on
         `agent_provision_plan` — the source of truth."""
 
-    @contextmanager
-    def launch(
-        self, plan: PlanT, *, skip_stale: bool = False
-    ) -> Generator[Bottle, None, None]:
-        """Template: optionally check for stale cached images, then delegate
-        to `_launch_impl`. Pass `skip_stale=True` to bypass the stale check
-        (used by the interactive CLI after the operator confirms)."""
-        if not skip_stale:
-            self._image_stale_checks(plan)
-        with self._launch_impl(plan) as bottle:
-            yield bottle
-
-    def _image_stale_checks(self, plan: PlanT) -> None:
-        """Raise StaleImageError if any cached image used by this plan is stale.
-        No-op default; backends override to call the shared `check_stale*`
-        helpers on their image/artifact timestamps."""
-
     @abstractmethod
-    def _launch_impl(self, plan: PlanT) -> AbstractContextManager[Bottle]:
+    def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
         """Build/run the bottle and yield a handle; tear down on exit."""
 
     def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:

bot_bottle/backend/docker/backend.py

@@ -85,11 +85,8 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
-    def _image_stale_checks(self, plan: DockerBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
     @contextmanager
-    def _launch_impl(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
+    def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:
             yield bottle
 

bot_bottle/backend/docker/compose.py

@@ -180,6 +180,10 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
 
     service: dict[str, Any] = {
         "image": SIDECAR_BUNDLE_IMAGE,
+        "build": {
+            "context": _REPO_DIR,
+            "dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
+        },
         "container_name": sidecar_bundle_container_name(plan.slug),
         "networks": {
             "internal": {"aliases": internal_aliases},
@@ -188,11 +192,6 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
         "environment": env,
         "volumes": volumes,
     }
-    if plan.spec.image_policy != "cached":
-        service["build"] = {
-            "context": _REPO_DIR,
-            "dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
-        }
     return service
 
 

bot_bottle/backend/docker/launch.py

@@ -41,8 +41,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import check_stale
+from ...log import info, warn
-from ...log import die, info, warn
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -64,7 +63,6 @@ from .compose import (
     write_compose_file,
 )
 from .egress import egress_tls_init
-from .sidecar_bundle import SIDECAR_BUNDLE_IMAGE
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
@@ -102,26 +100,12 @@ def launch(
         # Dockerfile. Sidecar images get built lazily by `docker compose
         # up` via the renderer's `build:` directives.
         committed = read_committed_image(plan.slug)
-        cached_policy = plan.spec.image_policy == "cached"
         if committed and docker_mod.image_exists(committed):
             info(f"using committed image {committed!r}")
             plan = dataclasses.replace(
                 plan,
                 agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
             )
-        elif cached_policy:
-            if not docker_mod.image_exists(plan.image):
-                die(
-                    f"cached agent image {plan.image!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            if not docker_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
-                die(
-                    f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
-                    "run without --cached-images to build it"
-                )
-            info(f"using cached agent image {plan.image!r}")
-            info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
         else:
             docker_mod.build_image(
                 plan.image, _REPO_DIR,
@@ -223,21 +207,3 @@ def launch(
         yield bottle
     finally:
         teardown()
-
-
-def stale_checks(plan: DockerBottlePlan) -> None:
-    """Raise StaleImageError if a cached image is older than the configured
-    threshold. Only runs when image_policy is 'cached'. Called by the backend
-    class's _image_stale_checks before _launch_impl starts any resources."""
-    if plan.spec.image_policy != "cached":
-        return
-    committed = read_committed_image(plan.slug)
-    if committed and docker_mod.image_exists(committed):
-        check_stale(f"agent image {committed!r}", docker_mod.image_created_at(committed))
-        return
-    for label, ref in [
-        ("agent image", plan.image),
-        ("sidecar image", SIDECAR_BUNDLE_IMAGE),
-    ]:
-        if docker_mod.image_exists(ref):
-            check_stale(f"{label} {ref!r}", docker_mod.image_created_at(ref))

bot_bottle/backend/docker/util.py

@@ -4,7 +4,6 @@ existence, and building images."""
 
 from __future__ import annotations
 
-from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -187,45 +186,6 @@ def image_id(ref: str) -> str:
     return r.stdout.strip()
 
 
-def image_created_at(ref: str) -> datetime:
-    """Return Docker's image Created timestamp as an aware UTC datetime."""
-    r = subprocess.run(
-        ["docker", "image", "inspect", "--format", "{{.Created}}", ref],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if r.returncode != 0:
-        die(
-            f"docker image inspect for {ref!r} failed: "
-            f"{(r.stderr or '').strip() or '<no stderr>'}"
-        )
-    raw = r.stdout.strip()
-    try:
-        return _parse_docker_timestamp(raw)
-    except ValueError:
-        die(f"docker image inspect for {ref!r} returned invalid Created timestamp: {raw!r}")
-
-
-def _parse_docker_timestamp(raw: str) -> datetime:
-    text = raw.strip()
-    if text.endswith("Z"):
-        text = text[:-1] + "+00:00"
-    dot = text.find(".")
-    if dot != -1:
-        tz_plus = text.find("+", dot)
-        tz_minus = text.find("-", dot)
-        tz_candidates = [pos for pos in (tz_plus, tz_minus) if pos != -1]
-        if tz_candidates:
-            tz_pos = min(tz_candidates)
-            frac = text[dot + 1:tz_pos]
-            text = text[:dot + 1] + frac[:6].ljust(6, "0") + text[tz_pos:]
-    dt = datetime.fromisoformat(text)
-    if dt.tzinfo is None:
-        dt = dt.replace(tzinfo=timezone.utc)
-    return dt.astimezone(timezone.utc)
-
-
 def save(ref: str, output: str) -> None:
     """`docker save REF -o OUTPUT`. Writes a tarball of the image
     layers + manifest to the host path. Used by smolmachines

bot_bottle/backend/macos_container/backend.py

@@ -67,11 +67,8 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
-    def _image_stale_checks(self, plan: MacosContainerBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
     @contextmanager
-    def _launch_impl(
+    def launch(
         self, plan: MacosContainerBottlePlan
     ) -> Generator[MacosContainerBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/macos_container/launch.py

@@ -32,7 +32,6 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import check_stale
 from ...log import die, info, warn
 from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
 from ...util import expand_tilde
@@ -150,53 +149,29 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
 
 
 def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
-    cached = plan.spec.image_policy == "cached"
+    container_mod.build_image(
+        SIDECAR_BUNDLE_IMAGE,
+        _REPO_DIR,
+        dockerfile=SIDECAR_BUNDLE_DOCKERFILE,
+    )
     committed = read_committed_image(plan.slug)
     if committed and container_mod.image_exists(committed):
         info(f"using committed image {committed!r}")
-        if not cached:
-            container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
         return dataclasses.replace(
             plan,
-            agent_provision=dataclasses.replace(plan.agent_provision, image=committed),
+            agent_provision=dataclasses.replace(
+                plan.agent_provision,
+                image=committed,
+            ),
         )
-    if cached:
+    container_mod.build_image(
-        if not container_mod.image_exists(plan.image):
+        plan.image,
-            die(
+        _REPO_DIR,
-                f"cached agent image {plan.image!r} not found; "
+        dockerfile=plan.dockerfile_path,
-                "run without --cached-images to build it"
+    )
-            )
-        if not container_mod.image_exists(SIDECAR_BUNDLE_IMAGE):
-            die(
-                f"cached sidecar image {SIDECAR_BUNDLE_IMAGE!r} not found; "
-                "run without --cached-images to build it"
-            )
-        info(f"using cached agent image {plan.image!r}")
-        info(f"using cached sidecar image {SIDECAR_BUNDLE_IMAGE!r}")
-        return plan
-    container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
-    container_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
     return plan
 
 
-def stale_checks(plan: MacosContainerBottlePlan) -> None:
-    """Raise StaleImageError if a cached image is older than the configured
-    threshold. Only runs when image_policy is 'cached'. Called by the backend
-    class's _image_stale_checks before _launch_impl starts any resources."""
-    if plan.spec.image_policy != "cached":
-        return
-    committed = read_committed_image(plan.slug)
-    if committed and container_mod.image_exists(committed):
-        check_stale(f"agent image {committed!r}", container_mod.image_created_at(committed))
-        return
-    for label, ref in [
-        ("agent image", plan.image),
-        ("sidecar image", SIDECAR_BUNDLE_IMAGE),
-    ]:
-        if container_mod.image_exists(ref):
-            check_stale(f"{label} {ref!r}", container_mod.image_created_at(ref))
-
-
 def _create_networks(
     internal_network: str,
     egress_network: str,

bot_bottle/backend/macos_container/util.py

@@ -10,7 +10,6 @@ import shutil
 import subprocess
 import tempfile
 import time
-from datetime import datetime, timezone
 from typing import Iterable
 
 from ...log import die, info
@@ -459,39 +458,6 @@ def image_id(ref: str) -> str:
     raise AssertionError("unreachable")
 
 
-def image_created_at(ref: str) -> datetime:
-    """Return the image creation timestamp as an aware UTC datetime.
-
-    Parses the `created` field from `container image inspect` JSON output."""
-    result = subprocess.run(
-        [_CONTAINER, "image", "inspect", ref],
-        capture_output=True,
-        text=True,
-        check=False,
-    )
-    if result.returncode != 0:
-        die(
-            f"container image inspect for {ref!r} failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}"
-        )
-    try:
-        data = json.loads(result.stdout or "{}")
-    except json.JSONDecodeError as exc:
-        die(f"container image inspect for {ref!r} returned malformed JSON: {exc}")
-    if isinstance(data, list) and data:
-        data = data[0]
-    if isinstance(data, dict):
-        value = data.get("created") or data.get("Created")
-        if isinstance(value, str) and value:
-            try:
-                ts = value.rstrip("Z")
-                return datetime.fromisoformat(ts).replace(tzinfo=timezone.utc)
-            except ValueError:
-                pass
-    die(f"container image inspect for {ref!r} did not include a creation timestamp")
-    raise AssertionError("unreachable")
-
-
 def save(ref: str, output: str) -> None:
     subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True)
 

bot_bottle/backend/smolmachines/backend.py

@@ -77,11 +77,8 @@ class SmolmachinesBottleBackend(
             stage_dir=stage_dir,
         )
 
-    def _image_stale_checks(self, plan: SmolmachinesBottlePlan) -> None:
-        _launch.stale_checks(plan)
-
     @contextmanager
-    def _launch_impl(
+    def launch(
         self, plan: SmolmachinesBottlePlan
     ) -> Generator[SmolmachinesBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/smolmachines/bottle_plan.py

@@ -1,10 +1,8 @@
 """SmolmachinesBottlePlan — concrete BottlePlan for the smolmachines
 backend (PRD 0023).
 
-Slug + bundle docker subnet / gateway / pinned IP + smolvm
+Slug + legacy bundle network coordinates + smolvm machine name +
-machine name + agent `.smolmachine` artifact + per-bottle guest
+agent `.smolmachine` artifact + per-bottle guest env."""
-env. Provisioning fields (CA cert path, prompt path, etc.) land
-in chunk 4."""
 
 from __future__ import annotations
 
@@ -23,9 +21,10 @@ class SmolmachinesBottlePlan(BottlePlan):
     `supervise_plan`, and `agent_provision` from BottlePlan."""
 
     slug: str
-    # Per-bottle docker subnet for the sidecar bundle container.
+    # Legacy per-bottle bundle network coordinates. These remain on
-    # The bundle runs at `bundle_ip` (always `.2`); the gateway is
+    # the plan while BundleLaunchSpec still carries the original shape,
-    # at `.1`. smolvm's TSI allowlist is set to `bundle_ip/32`.
+    # but the smolmachines launch path exposes the sidecar VM through
+    # host-loopback forwarders instead of a Docker bridge IP.
     bundle_subnet: str
     bundle_gateway: str
     bundle_ip: str
@@ -36,22 +35,10 @@ class SmolmachinesBottlePlan(BottlePlan):
     # `--smolfile` is mutually exclusive with `--from`, and
     # `--from` is the path that avoids the registry-pull race).
     guest_env: dict[str, str]
-    # Inner Plans for the sidecar bundle daemons. The same shape the
+    # Agent-side endpoints. Empty at prepare time; launch populates
-    # docker backend uses — same `.prepare()` calls produced
+    # these after sidecar VM bringup via `dataclasses.replace`.
-    # them — but our launch step doesn't populate the
+    # Format: a `host:port` for git-gate (insteadOf URL prefix) +
-    # docker-specific network fields (internal_network,
+    # full URLs for proxy / supervise.
-    # egress_network) because the smolmachines bundle isn't on
-    # docker's `--internal` + egress bridge topology; it's on a
-    # per-bottle bridge with a pinned IP. The unused fields stay
-    # at their dataclass defaults.
-    # Agent-side endpoints. On Docker Desktop the docker bridge
-    # IPs aren't reachable from the smolvm guest (TSI uses macOS
-    # networking; docker container IPs live in the daemon's VM),
-    # so the agent dials the bundle via host loopback +
-    # docker-published random ports. Empty at prepare time;
-    # launch populates these after bundle bringup via
-    # `dataclasses.replace`. Format: a `host:port` for git-gate
-    # (insteadOf URL prefix) + full URLs for proxy / supervise.
     agent_proxy_url: str = ""
     agent_git_gate_host: str = ""
     agent_supervise_url: str = ""

bot_bottle/backend/smolmachines/launch.py

@@ -1,12 +1,9 @@
-"""End-to-end launch flow for the smolmachines backend
+"""End-to-end launch flow for the smolmachines backend.
-(PRD 0023 chunks 2d + 4b).
 
-Brings up the per-bottle docker bridge + sidecar bundle (with
+Builds the sidecar bundle smolmachine, starts it as a sidecar VM
-real daemons + their config files), creates + starts the smolvm
+with real daemons + their config files, creates + starts the agent
-guest pointed at the bundle's pinned IP via TSI's
+smolVM, yields a `SmolmachinesBottle` handle, and tears everything
-`--allow-cidr <bundle-ip>/32` allowlist, yields a
+down on context exit.
-`SmolmachinesBottle` handle, tears everything down on context
-exit.
 
 The bundle's daemons consume the inner Plans the docker backend
 already produces: egress reads routes + CAs from the EgressPlan.
@@ -45,8 +42,7 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...image_cache import check_stale_path
+from ...log import info, warn
-from ...log import die, info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
@@ -73,9 +69,8 @@ _SMOLMACHINE_CACHE_DIR = Path.home() / ".cache" / "bot-bottle" / "smolmachines"
 
 
 # Container-internal listening ports for each bundle daemon. The
-# bundle publishes each one on a random host loopback port (see
+# sidecar VM publishes each one on a random host loopback port, and
-# `_bundle.start_bundle`), and `_bundle.bundle_host_port` looks
+# the launch flow wraps those raw ports with per-bottle forwarders.
-# them up post-start.
 _GIT_HTTP_PORT = 9420
 _SUPERVISE_PORT = SUPERVISE_PORT
 
@@ -183,13 +178,10 @@ def _start_bundle(
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
-    if _image_policy(plan) == "cached":
+    artifact = _ensure_smolmachine(
-        artifact = _cached_smolmachine(bundle_spec.image, label="sidecar")
+        bundle_spec.image,
-    else:
+        dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-        artifact = _ensure_smolmachine(
+    )
-            bundle_spec.image,
-            dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
-        )
     launch = _bundle.start_bundle_vm(
         bundle_spec,
         from_path=artifact,
@@ -473,9 +465,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
             info(f"using committed smolmachine {str(committed_path)!r}")
             return committed_path
 
-    if _image_policy(plan) == "cached":
-        return _cached_smolmachine(plan.agent_image, label="agent")
-
     # Build the agent image and pack it into a `.smolmachine`
     # artifact (or hit the per-Dockerfile-digest cache). Runs here,
     # not in prepare, so the docker-build output doesn't garble the
@@ -486,60 +475,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
     )
 
 
-def stale_checks(plan: SmolmachinesBottlePlan) -> None:
-    """Raise StaleImageError if a cached smolmachine artifact is older than the
-    configured threshold. Only runs when image_policy is 'cached'. Checks the
-    committed agent artifact first, then the cache-keyed agent and sidecar
-    artifacts. Called by the backend class's _image_stale_checks before any
-    resources are allocated."""
-    if _image_policy(plan) != "cached":
-        return
-    committed = read_committed_image(plan.slug)
-    if committed:
-        committed_path = Path(committed)
-        if committed_path.is_file():
-            check_stale_path("agent smolmachine artifact", committed_path)
-    elif docker_mod.image_exists(plan.agent_image):
-        digest = docker_mod.image_id(plan.agent_image).split(":", 1)[-1][:16]
-        artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-        if artifact.is_file():
-            check_stale_path("agent smolmachine artifact", artifact)
-    sidecar_image = _bundle.SIDECAR_BUNDLE_IMAGE
-    if docker_mod.image_exists(sidecar_image):
-        digest = docker_mod.image_id(sidecar_image).split(":", 1)[-1][:16]
-        sidecar_artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-        if sidecar_artifact.is_file():
-            check_stale_path("sidecar smolmachine artifact", sidecar_artifact)
-
-
-def _image_policy(plan: object) -> str:
-    spec = getattr(plan, "spec", None)
-    return str(getattr(spec, "image_policy", "fresh"))
-
-
-def _cached_smolmachine(image_ref: str, *, label: str) -> Path:
-    """Return the cached smolmachine artifact for the current local image.
-
-    This is intentionally buildless: it inspects the existing local Docker
-    image ID and looks for the artifact keyed by that ID. If either side is
-    missing, the caller must use the fresh path to build/pack it.
-    """
-    if not docker_mod.image_exists(image_ref):
-        die(
-            f"cached {label} image {image_ref!r} not found; "
-            "run without --cached-images to build it"
-        )
-    digest = docker_mod.image_id(image_ref).split(":", 1)[-1][:16]
-    sidecar = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
-    if not sidecar.is_file():
-        die(
-            f"cached {label} smolmachine artifact for {image_ref!r} not found; "
-            "run without --cached-images to build it"
-        )
-    info(f"using cached {label} smolmachine {str(sidecar)!r}")
-    return sidecar
-
-
 def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:
     """Build the agent docker image and convert it into a
     `.smolmachine` artifact, caching the result under

bot_bottle/backend/smolmachines/resolve_plan.py

@@ -56,13 +56,11 @@ def resolve_plan(
     git_gate_plan: GitGatePlan,
     stage_dir: Path,
 ) -> SmolmachinesBottlePlan:
-    """Materialize the smolmachines plan. The bundle's docker
+    """Materialize the smolmachines plan. The agent `.smolmachine`
-    subnet + pinned IP are derived from the slug; the agent's
+    artifact is built (or cache-hit) here so launch's
-    `.smolmachine` artifact is built (or cache-hit) here so
+    `machine create --from` boots without a registry pull. Per-bottle
-    launch's `machine create --from` boots without a registry
+    guest env lands on the plan for launch to pass straight through
-    pull. Per-bottle guest env + the TSI allow_cidrs land on the
+    to `machine create` flags."""
-    plan for launch to pass straight through to
-    `machine create` flags."""
 
     # ==== smolmachines specific setup ====
     subnet, gateway, bundle_ip = smolmachines_bundle_subnet(slug)

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -1,36 +1,18 @@
-"""Per-bottle sidecar bundle bringup for the smolmachines backend
+"""Per-bottle sidecar bundle bringup for the smolmachines backend.
-(PRD 0023).
 
-Two docker resources per bottle live here:
+The sidecar bundle runs as its own smolVM. The agent VM reaches
-
+bundle daemons through host-loopback ports published by that sidecar
-  - **A dedicated bridge network**, subnet derived from the slug.
+VM and wrapped by per-bottle address-bound forwarders."""
-    The bundle container gets a pinned IP at `<subnet>.2` so the
-    smolvm guest's TSI allowlist (`<bundle-ip>/32`) has a stable
-    target. Without pinning, we'd have to inspect the container's
-    assigned IP after start and feed it back into the Smolfile
-    — a race we can sidestep with `--ip`.
-
-  - **The bundle container itself**, running the PRD 0024 bundle
-    image (`bot-bottle-sidecars:latest` by default). Same
-    image, same daemons, same daemon-private env / bind-mounts
-    as the docker backend.
-
-This module ships the lifecycle primitives only — create
-network, start bundle, stop bundle, remove network — wrapped
-around `subprocess.run(["docker", ...])`. Wiring them into the
-launch flow + populating the `BundleLaunchSpec` from the inner
-Plans (EgressPlan, …) lands in chunk 2d."""
 
 from __future__ import annotations
 
 import os
 import socket
-import subprocess
 from dataclasses import dataclass, field
 from pathlib import Path
 from typing import Sequence
 
-from ...log import die, warn
+from ...log import warn
 from ..docker import util as docker_mod
 from ..docker.sidecar_bundle import (
     SIDECAR_BUNDLE_DOCKERFILE,
@@ -78,10 +60,8 @@ class BundleLaunchSpec:
     # supervisor inside the bundle reads it to skip
     # bottle-irrelevant daemons (e.g. supervise=False bottles).
     daemons_csv: str = "egress"
-    # Plain "KEY=VALUE" strings + "KEY" bare names (the bare-name
+    # Plain "KEY=VALUE" strings + "KEY" bare names. Bare names inherit
-    # form inherits the value from the docker-run subprocess env,
+    # from the host env passed to the sidecar VM launch.
-    # matching the docker backend's compose-up secret-forwarding
-    # pattern).
     environment: Sequence[str] = field(default_factory=tuple)
     # (host_path, container_path, read_only) bind mounts.
     volumes: Sequence[tuple[str, str, bool]] = field(default_factory=tuple)
@@ -182,138 +162,3 @@ def stop_bundle_vm(slug: str) -> None:
         _smolvm.machine_delete(name)
     except _smolvm.SmolvmError as exc:
         warn(f"smolvm machine delete {name} failed: {exc}")
-
-
-def create_bundle_network(network_name: str, subnet: str, gateway: str) -> None:
-    """`docker network create` with an explicit subnet + gateway
-    so the bundle's `--ip` lands on the address the Smolfile's
-    TSI allowlist points at. Idempotent on the caller's side —
-    `start_bundle` catches the "network exists" error and treats
-    it as success (chunk-2d teardown is paired with each create).
-    """
-    result = subprocess.run(
-        ["docker", "network", "create",
-         "--subnet", subnet, "--gateway", gateway,
-         network_name],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        # Already-exists is fine on a resume path; everything else
-        # is fatal — the bundle won't have an addressable network.
-        if "already exists" in (result.stderr or "").lower():
-            return
-        die(
-            f"docker network create {network_name} failed: "
-            f"{(result.stderr or '').strip()}"
-        )
-
-
-def remove_bundle_network(network_name: str) -> None:
-    """Idempotent: a missing network returns success."""
-    result = subprocess.run(
-        ["docker", "network", "rm", network_name],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode == 0:
-        return
-    if "no such network" in (result.stderr or "").lower():
-        return
-    # Network with attached containers is the common non-fatal
-    # case during a partial teardown — warn but don't die.
-    warn(
-        f"docker network rm {network_name} failed: "
-        f"{(result.stderr or '').strip()}"
-    )
-
-
-def start_bundle(spec: BundleLaunchSpec, *,
-                 env: dict[str, str] | None = None) -> None:
-    """Bring the bundle container up on the per-bottle bridge with
-    the pinned IP. Argv is built deterministically from `spec`;
-    `env` is the host subprocess env (forwarded values for any
-    bare-name entries in `spec.environment`)."""
-    container = bundle_container_name(spec.slug)
-    argv = [
-        "docker", "run",
-        "--name", container,
-        "--detach",
-        "--rm",
-        "--network", spec.network_name,
-        "--ip", spec.bundle_ip,
-        "-e", f"BOT_BOTTLE_SIDECAR_DAEMONS={spec.daemons_csv}",
-    ]
-    for entry in spec.environment:
-        argv += ["-e", entry]
-    for host_path, container_path, read_only in spec.volumes:
-        suffix = ":ro" if read_only else ""
-        argv += ["-v", f"{host_path}:{container_path}{suffix}"]
-    # Loopback-only host port-forwards — the smolvm guest's TSI
-    # uses macOS networking, and macOS loopback is the only host
-    # surface that round-trips into Docker Desktop's daemon VM.
-    # Binds to the per-bottle alias so TSI's IP-only allowlist
-    # narrows reachability to this bottle's bundle only.
-    for port in spec.ports_to_publish:
-        argv += ["-p", f"{spec.publish_host_ip}::{port}"]
-    argv.append(spec.image)
-    result = subprocess.run(
-        argv, capture_output=True, text=True,
-        env=dict(env) if env is not None else None, check=False,
-    )
-    if result.returncode != 0:
-        die(
-            f"docker run for bundle {container} failed: "
-            f"{(result.stderr or '').strip()}"
-        )
-
-
-def bundle_host_port(
-    slug: str, container_port: int, *, host_ip: str = "127.0.0.1",
-) -> int:
-    """`docker port <bundle> <container_port>/tcp` → the random
-    host-side port docker assigned for the binding on `host_ip`.
-    Called after `start_bundle` on each container port listed in
-    `BundleLaunchSpec.ports_to_publish` so the launch step can
-    build the agent's HTTPS_PROXY / GIT_GATE / SUPERVISE URLs in
-    `<host_ip>:<host port>` form."""
-    container = bundle_container_name(slug)
-    result = subprocess.run(
-        ["docker", "port", container, f"{container_port}/tcp"],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode != 0:
-        die(
-            f"docker port {container} {container_port}/tcp failed: "
-            f"{(result.stderr or '').strip() or '<no stderr>'}"
-        )
-    # Each line looks like `127.0.0.16:54321` — one per address
-    # family / host IP. Match on the expected host_ip prefix so
-    # bottles bound to per-bottle aliases pick the right line.
-    for raw in (result.stdout or "").splitlines():
-        line = raw.strip()
-        if line.startswith(f"{host_ip}:"):
-            _, _, port_str = line.rpartition(":")
-            try:
-                return int(port_str)
-            except ValueError:
-                die(f"unexpected `docker port` output: {line!r}")
-    die(
-        f"no port mapping on {host_ip} for {container} "
-        f"{container_port}/tcp; got: {(result.stdout or '').strip()!r}"
-    )
-
-
-def stop_bundle(slug: str) -> None:
-    """Idempotent: a missing container returns success."""
-    container = bundle_container_name(slug)
-    result = subprocess.run(
-        ["docker", "rm", "-f", container],
-        capture_output=True, text=True, check=False,
-    )
-    if result.returncode == 0:
-        return
-    if "no such container" in (result.stderr or "").lower():
-        return
-    warn(
-        f"docker rm -f {container} failed: "
-        f"{(result.stderr or '').strip()}"
-    )

bot_bottle/cli/start.py

@@ -36,7 +36,6 @@ from ..bottle_state import (
     is_preserved,
     mark_preserved,
 )
-from ..image_cache import StaleImageError
 from ..log import info, die
 from ..manifest import Manifest, ManifestIndex
 from ._common import PROG, USER_CWD, read_tty_line
@@ -64,14 +63,6 @@ def cmd_start(argv: list[str]) -> int:
             "skip all prompts. For orchestrators, CI, and webhooks."
         ),
     )
-    parser.add_argument(
-        "--cached-images",
-        action="store_true",
-        help=(
-            "quickstart with existing local agent and sidecar images; "
-            "only valid with --headless"
-        ),
-    )
     parser.add_argument(
         "--bottle",
         action="append",
@@ -104,8 +95,6 @@ def cmd_start(argv: list[str]) -> int:
         help="agent name defined in bot-bottle.json (omit to pick interactively)",
     )
     args = parser.parse_args(argv)
-    if args.cached_images and not args.headless:
-        die("--cached-images is only supported with --headless")
 
     dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
 
@@ -153,10 +142,6 @@ def cmd_start(argv: list[str]) -> int:
     label, color = tui.name_color_modal(default_label=agent_name)
     label, color = _resolve_unique_label(label, color)
 
-    image_policy = _select_image_policy()
-    if image_policy is None:
-        return 0
-
     spec = BottleSpec(
         manifest=manifest,
         agent_name=agent_name,
@@ -165,7 +150,6 @@ def cmd_start(argv: list[str]) -> int:
         label=label,
         color=color,
         bottle_names=bottle_names,
-        image_policy=image_policy,
     )
     return _launch_bottle(
         spec,
@@ -226,7 +210,6 @@ def _start_headless(
         color=args.color or "",
         bottle_names=bottle_names,
         headless=True,
-        image_policy="cached" if args.cached_images else "fresh",
     )
     return _launch_bottle(
         spec,
@@ -407,13 +390,6 @@ def _text_prompt_yes() -> bool:
     return reply in ("y", "Y", "yes", "YES")
 
 
-def _select_image_policy() -> str | None:
-    return tui.filter_select(
-        ["fresh", "cached"],
-        title="Select image startup mode",
-    )
-
-
 def _text_render_preflight():
     def _render(plan: DockerBottlePlan, backend_name: str) -> None:
         print(file=sys.stderr)
@@ -556,38 +532,30 @@ def _launch_bottle(
             return 0
 
         backend = get_bottle_backend(backend_name)
-        skip_stale = False
+        with backend.launch(plan) as bottle:
-        while True:
+            agent_provider_template = getattr(plan, "agent_provider_template", "claude")
-            try:
+            extra_args: tuple[str, ...] = ()
-                with backend.launch(plan, skip_stale=skip_stale) as bottle:
+            if headless_prompt_text:
-                    agent_provider_template = getattr(plan, "agent_provider_template", "claude")
+                extra_args = tuple(
-                    extra_args: tuple[str, ...] = ()
+                    get_provider(agent_provider_template).headless_prompt(
-                    if headless_prompt_text:
+                        headless_prompt_text
-                        extra_args = tuple(
-                            get_provider(agent_provider_template).headless_prompt(
-                                headless_prompt_text
-                            )
-                        )
-                    exit_code = attach_agent(
-                        bottle,
-                        agent_provider_template=agent_provider_template,
-                        startup_args=plan.agent_provision.startup_args + extra_args,
                     )
-                    info(
+                )
-                        f"session ended (exit {exit_code}); "
+            exit_code = attach_agent(
-                        f"container {bottle.name} will be removed"
+                bottle,
-                    )
+                agent_provider_template=agent_provider_template,
-                    if agent_provider_template == "claude":
+                startup_args=plan.agent_provision.startup_args + extra_args,
-                        capture_claude_session_state(identity, exit_code)
+            )
-                break
+            info(
-            except StaleImageError as exc:
+                f"session ended (exit {exit_code}); "
-                if assume_yes:
+                f"container {bottle.name} will be removed"
-                    die(str(exc))
+            )
-                sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
+            # While the container is still alive: always snapshot the
-                sys.stderr.flush()
+            # transcript and — if the agent exited non-zero — mark
-                if read_tty_line() not in ("y", "Y", "yes", "YES"):
+            # the state for preservation. This picks up crashes /
-                    break
+            # Ctrl-Cs / OOM kills before cleanup removes the state dir.
-                skip_stale = True
+            if agent_provider_template == "claude":
+                capture_claude_session_state(identity, exit_code)
         return 0
     finally:
         # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount

bot_bottle/config_store.py

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

bot_bottle/image_cache.py

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

bot_bottle/store_manager.py

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

tests/integration/test_smolmachines_bundle_bringup.py

@@ -1,111 +0,0 @@
-"""Integration: PRD 0023 chunk 2c — bundle bringup on a per-bottle
-docker bridge with the pinned IP.
-
-End-to-end against the real docker daemon. Brings up just the
-sidecar bundle on its own bridge, confirms the container lands at
-the pinned IP, then tears down. Skipped under act_runner (docker
-socket mount topology breaks bridge visibility) and when the
-bundle image isn't available.
-
-Full launch flow (smolvm + bundle + provisioning + the
-localhost-reach / egress-port-bypass probes) lives in chunk 2d."""
-
-from __future__ import annotations
-
-import os
-import subprocess
-import time
-import unittest
-
-from bot_bottle.backend.smolmachines.sidecar_bundle import (
-    BundleLaunchSpec,
-    bundle_container_name,
-    bundle_network_name,
-    create_bundle_network,
-    remove_bundle_network,
-    start_bundle,
-    stop_bundle,
-)
-from tests._docker import skip_unless_docker
-
-
-@skip_unless_docker()
-@unittest.skipIf(
-    os.environ.get("GITEA_ACTIONS") == "true",
-    "skipped under act_runner: docker socket mount topology breaks "
-    "in-process visibility of networks created on the host daemon",
-)
-class TestBundleBringup(unittest.TestCase):
-    def setUp(self):
-        self.slug = f"cb-test-bundle-{os.getpid()}-{int(time.time())}"
-        self.network = bundle_network_name(self.slug)
-        self.container = bundle_container_name(self.slug)
-
-    def tearDown(self):
-        stop_bundle(self.slug)
-        remove_bundle_network(self.network)
-
-    def _bundle_image_built(self) -> bool:
-        """The bundle image (`bot-bottle-sidecars:latest`) is
-        built lazily by the docker backend's compose. If a
-        smolmachines-only operator hasn't run the docker backend
-        first, the image won't exist locally. Skip rather than
-        fail."""
-        r = subprocess.run(
-            ["docker", "image", "inspect", "bot-bottle-sidecars:latest"],
-            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
-            check=False,
-        )
-        return r.returncode == 0
-
-    def test_create_network_then_start_bundle_pins_ip(self):
-        if not self._bundle_image_built():
-            self.skipTest(
-                "bot-bottle-sidecars:latest not built; run a docker "
-                "bottle first or `docker build -f Dockerfile.sidecars .`"
-            )
-
-        # Pick a subnet unlikely to collide on the host. Last
-        # octet of the slug hash isn't deterministic across runs;
-        # we hardcode a high octet (.211) that the docker default
-        # bridges almost never use.
-        subnet = "192.168.211.0/24"
-        gateway = "192.168.211.1"
-        bundle_ip = "192.168.211.2"
-
-        create_bundle_network(self.network, subnet, gateway)
-
-        spec = BundleLaunchSpec(
-            slug=self.slug,
-            network_name=self.network,
-            subnet=subnet,
-            gateway=gateway,
-            bundle_ip=bundle_ip,
-            # Empty daemons_csv → init exits "no daemons selected"
-            # immediately. We just need the container to land on
-            # the network at the right IP before it exits.
-            daemons_csv="",  # empty → init exits "no daemons selected"
-        )
-        start_bundle(spec)
-
-        # Inspect the container's IP on the per-bottle network.
-        r = subprocess.run(
-            ["docker", "inspect",
-             "--format",
-             "{{(index .NetworkSettings.Networks \"" + self.network + "\").IPAddress}}",
-             self.container],
-            capture_output=True, text=True, check=False,
-        )
-        # Container may have exited (no daemons selected → exit 0).
-        # The inspect still works on exited containers as long as
-        # `--rm` hasn't fired yet, which is a race. Even if it has,
-        # the launch succeeded — the container existed, on the
-        # right network, at the right IP. We don't fail here on
-        # missing inspect.
-        if r.returncode == 0 and r.stdout.strip():
-            self.assertEqual(bundle_ip, r.stdout.strip(),
-                             f"bundle landed at wrong IP: {r.stdout!r}")
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/integration/test_smolmachines_launch.py

@@ -1,10 +1,9 @@
-"""Integration: PRD 0023 chunk 2d — end-to-end launch + exec
+"""Integration: end-to-end smolmachines launch + exec round trip.
-round trip + the acceptance probes.
 
-The smoke confirms the launch flow (per-bottle docker bridge →
+The smoke confirms the launch flow (sidecar bundle smolVM →
-sidecar bundle with host-loopback published ports → smolvm guest
+host-loopback forwarders → agent smolVM with TSI allowlist → exec)
-with TSI allowlist → exec) plumbs together end to end. The probes confirm the
+plumbs together end to end. The probes confirm the security
-security properties the design pivot was about:
+properties the design pivot was about:
 
   - **localhost-reach probe** — guest tries to dial a service
     bound on the host's `127.0.0.1`. TSI's per-bottle loopback
@@ -14,13 +13,6 @@ security properties the design pivot was about:
     the injected `HTTPS_PROXY`/`HTTP_PROXY` URL on the per-bottle
     loopback alias, while direct egress with proxy vars unset fails.
 
-  - **egress-port-bypass probe** — guest tries to dial
-    `<bundle-ip>:9099` (egress's port). TSI permits the IP but
-    the bundle's egress daemon binds `127.0.0.1` inside its
-    container, so the connect refuses at the socket level. The
-    bind-address mitigation is what closes TSI's port-granularity
-    gap.
-
 Gated on macOS/Linux + smolvm + docker + not GITEA_ACTIONS — the
 runner can't host libkrun-backed VMs."""
 
@@ -114,8 +106,8 @@ class TestSmolmachinesLaunch(unittest.TestCase):
 
     def test_localhost_reach_probe(self):
         # Agent dials a 127.0.0.1 service on the host. TSI's
-        # allowlist contains only <bundle-ip>/32, so this must
+        # allowlist contains only the per-bottle loopback alias, so
-        # refuse. We use a port unlikely to be bound on the host
+        # this must refuse. We use a port unlikely to be bound on the host
         # (high-numbered) so we're confirming TSI refusal, not
         # just "no service listening."
         r = self.bottle.exec(
@@ -196,28 +188,6 @@ class TestSmolmachinesLaunch(unittest.TestCase):
         self.assertEqual(0, r.returncode, msg=r.stderr)
         self.assertEqual(_AGENT_PROMPT, r.stdout.rstrip("\n"))
 
-    def test_egress_port_bypass_probe(self):
-        # Agent dials <bundle-ip>:9099 (egress's port). TSI
-        # permits the IP, but egress will bind 127.0.0.1:9099
-        # inside the bundle in chunk 3, so the connect refuses
-        # at the socket level. NOTE: in chunk 2d the bundle's
-        # daemons aren't running (daemons_csv=""), so nothing
-        # is listening on :9099 anyway — this test asserts the
-        # connect fails, which is the property chunk 3 will
-        # preserve once egress is actually running.
-        r = self.bottle.exec(
-            "env -u HTTPS_PROXY -u HTTP_PROXY -u https_proxy -u http_proxy "
-            f"curl -s --show-error --max-time 3 http://{self.plan.bundle_ip}:9099 "
-            "2>&1 || true"
-        )
-        self.assertTrue(
-            "refused" in r.stdout.lower()
-            or "timed out" in r.stdout.lower()
-            or "unreachable" in r.stdout.lower()
-            or "failed" in r.stdout.lower(),
-            f"expected egress port refusal; got: {r.stdout!r}",
-        )
-
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_cli_start_headless.py

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

tests/unit/test_cli_start_selector.py

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

tests/unit/test_cli_start_stale.py

@@ -1,164 +0,0 @@
-"""Unit: _launch_bottle StaleImageError handling.
-
-Exercises the while-True / try-except loop around backend.launch:
-  - headless mode → die on stale
-  - interactive mode, user declines → stop without launching
-  - interactive mode, user confirms → retry with skip_stale=True
-"""
-
-from __future__ import annotations
-
-import io
-import tempfile
-import unittest
-from types import SimpleNamespace
-from typing import Any, cast
-from unittest.mock import MagicMock, patch
-
-from bot_bottle.image_cache import StaleImageError
-from bot_bottle.log import Die
-
-
-def _fake_plan() -> Any:
-    provision = SimpleNamespace(startup_args=())
-    return cast(Any, SimpleNamespace(
-        agent_provision=provision,
-        agent_provider_template="claude",
-        slug="dev-abc",
-    ))
-
-
-def _stale_cm() -> MagicMock:
-    """Return a context-manager mock whose __enter__ raises StaleImageError."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(side_effect=StaleImageError("image is 5 day(s) old"))
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
-def _ok_cm(bottle: Any) -> MagicMock:
-    """Return a context-manager mock that yields `bottle`."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(return_value=bottle)
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
-class TestLaunchBottleStaleHandling(unittest.TestCase):
-    def setUp(self) -> None:
-        self._tmp = tempfile.mkdtemp(prefix="cli-stale-test.")
-
-    def _spec(self) -> Any:
-        from bot_bottle.backend import BottleSpec
-        from bot_bottle.manifest import ManifestIndex
-        idx = ManifestIndex.from_json_obj({
-            "bottles": {"dev": {}},
-            "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
-        })
-        return BottleSpec(
-            manifest=idx,
-            agent_name="demo",
-            copy_cwd=False,
-            user_cwd=self._tmp,
-            identity="dev-abc",
-        )
-
-    def _run_launch(self, **patch_kwargs: Any) -> int:
-        import bot_bottle.cli.start as start_mod
-        spec = self._spec()
-        with patch.object(start_mod, "prepare_with_preflight",
-                          return_value=(_fake_plan(), "dev-abc")), \
-             patch.object(start_mod, "settle_state"), \
-             patch.object(start_mod, "info"):
-            return start_mod._launch_bottle(
-                spec,
-                dry_run=False,
-                backend_name="docker",
-                **patch_kwargs,
-            )
-
-    def test_headless_stale_calls_die(self) -> None:
-        """In headless mode (assume_yes=True), a StaleImageError must call die()."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "die", side_effect=Die()):
-            with self.assertRaises(Die):
-                self._run_launch(assume_yes=True)
-
-    def test_interactive_user_declines_stops_loop(self) -> None:
-        """Interactive user answering 'n' → loop exits without a second launch."""
-        import bot_bottle.cli.start as start_mod
-
-        backend_mock = MagicMock()
-        backend_mock.launch.return_value = _stale_cm()
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="n"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        # launch was called once (stale), then the user declined → no second call.
-        backend_mock.launch.assert_called_once()
-
-    def test_interactive_user_confirms_retries_with_skip_stale(self) -> None:
-        """Interactive user answering 'y' → second launch with skip_stale=True."""
-        import bot_bottle.cli.start as start_mod
-
-        bottle_mock = MagicMock()
-        bottle_mock.name = "dev-abc"
-
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
-        backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="y"), \
-             patch.object(start_mod, "attach_agent", return_value=0), \
-             patch.object(start_mod, "capture_claude_session_state"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        # First call: skip_stale=False → stale. Second call: skip_stale=True → ok.
-        self.assertEqual(2, backend_mock.launch.call_count)
-        calls = backend_mock.launch.call_args_list
-        self.assertFalse(calls[0].kwargs.get("skip_stale", False))
-        self.assertTrue(calls[1].kwargs.get("skip_stale", False))
-
-    def test_interactive_yes_uppercase_also_accepted(self) -> None:
-        """'Y' or 'YES' should also be accepted as confirmation."""
-        import bot_bottle.cli.start as start_mod
-
-        bottle_mock = MagicMock()
-        bottle_mock.name = "dev-abc"
-
-        def launch_side_effect(plan: Any, *, skip_stale: bool = False) -> Any:
-            if not skip_stale:
-                return _stale_cm()
-            return _ok_cm(bottle_mock)
-
-        backend_mock = MagicMock()
-        backend_mock.launch.side_effect = launch_side_effect
-
-        with patch.object(start_mod, "get_bottle_backend", return_value=backend_mock), \
-             patch.object(start_mod, "read_tty_line", return_value="YES"), \
-             patch.object(start_mod, "attach_agent", return_value=0), \
-             patch.object(start_mod, "capture_claude_session_state"), \
-             patch("sys.stderr", new_callable=io.StringIO):
-            rc = self._run_launch(assume_yes=False)
-
-        self.assertEqual(0, rc)
-        self.assertEqual(2, backend_mock.launch.call_count)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_compose.py

@@ -118,7 +118,6 @@ def _plan(
     with_egress: bool = False,
     supervise: bool = False,
     canary: bool = False,
-    image_policy: str = "fresh",
 ) -> DockerBottlePlan:
     """Build a fully-resolved DockerBottlePlan. Toggles cover the
     matrix the renderer's conditional-service logic branches on."""
@@ -149,7 +148,6 @@ def _plan(
         agent_name="demo",
         copy_cwd=False,
         user_cwd="/tmp/x",
-        image_policy=image_policy,
     )
     return DockerBottlePlan(
         spec=spec,
@@ -267,8 +265,7 @@ class TestAgentAlwaysPresent(unittest.TestCase):
     def test_agent_depends_only_on_sidecars(self):
         # Bundle shape: the init supervisor owns intra-bundle daemon
         # ordering, so the agent waits on the bundle container alone.
-        cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
+        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
-        for kwargs in cases:
             with self.subTest(**kwargs):
                 s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
                 self.assertEqual(["sidecars"], s["depends_on"])
@@ -303,11 +300,6 @@ class TestSidecarBundleShape(unittest.TestCase):
         self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
         self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
 
-    def test_cached_policy_omits_bundle_build(self):
-        sc = self._render(image_policy="cached")["services"]["sidecars"]
-        self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
-        self.assertNotIn("build", sc)
-
     def test_bundle_container_name_uses_sidecars_prefix(self):
         sc = self._render()["services"]["sidecars"]
         self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])

tests/unit/test_config_store.py

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

tests/unit/test_docker_launch_committed_image.py

@@ -3,7 +3,6 @@
 from __future__ import annotations
 
 import contextlib
-import dataclasses
 import io
 import tempfile
 import unittest
@@ -17,7 +16,6 @@ from bot_bottle.backend.docker import launch as launch_mod
 from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 
 
@@ -182,24 +180,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
         built = self._run_launch(plan, committed_tag=None)
         self.assertEqual([_DEFAULT_IMAGE], built)
 
-    def test_cached_images_skip_build_when_present(self) -> None:
-        base = _plan(self._tmp)
-        plan = dataclasses.replace(
-            base,
-            spec=dataclasses.replace(base.spec, image_policy="cached"),
-        )
-        built = self._run_launch(plan, committed_tag=None, image_present=True)
-        self.assertEqual([], built)
-
-    def test_cached_images_die_when_agent_missing(self) -> None:
-        base = _plan(self._tmp)
-        plan = dataclasses.replace(
-            base,
-            spec=dataclasses.replace(base.spec, image_policy="cached"),
-        )
-        with self.assertRaises(Die):
-            self._run_launch(plan, committed_tag=None, image_present=False)
-
     def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
         plan = _plan(self._tmp)
         built = self._run_launch(

tests/unit/test_docker_util_image.py

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

tests/unit/test_image_cache.py

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

tests/unit/test_macos_container_launch.py

@@ -2,7 +2,6 @@
 
 from __future__ import annotations
 
-import dataclasses
 import unittest
 import tempfile
 from pathlib import Path
@@ -16,7 +15,6 @@ from bot_bottle.backend.macos_container import launch
 from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
 from bot_bottle.egress import EgressPlan
 from bot_bottle.git_gate import GitGatePlan
-from bot_bottle.log import Die
 from bot_bottle.manifest import ManifestIndex
 
 _MANIFEST = ManifestIndex.from_json_obj({
@@ -83,7 +81,7 @@ def _plan(
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(image_policy="fresh"),
+        spec=SimpleNamespace(),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
@@ -294,7 +292,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
 
 def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
     return MacosContainerBottlePlan(
-        spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
+        spec=cast(BottleSpec, SimpleNamespace()),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -367,82 +365,5 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
         self.assertEqual("bot-bottle-agent:latest", calls[1][0])
 
 
-class TestMacosContainerLaunchCachedImages(unittest.TestCase):
-    def setUp(self):
-        self._tmp = tempfile.TemporaryDirectory()
-        self.stage_dir = Path(self._tmp.name)
-
-    def tearDown(self):
-        self._tmp.cleanup()
-
-    def _cached_plan(self) -> MacosContainerBottlePlan:
-        return dataclasses.replace(
-            _build_plan(self.stage_dir),
-            spec=cast(BottleSpec, SimpleNamespace(image_policy="cached")),
-        )
-
-    def test_cached_mode_dies_when_agent_image_missing(self) -> None:
-        plan = self._cached_plan()
-
-        def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
-            return False  # Nothing cached.
-
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", side_effect=image_exists,
-        ), patch.object(
-            launch, "die", side_effect=Die(),
-        ):
-            with self.assertRaises(Die):
-                launch._build_images(plan)
-
-    def test_cached_mode_dies_when_sidecar_image_missing(self) -> None:
-        plan = self._cached_plan()
-
-        def image_exists(ref: str) -> bool:
-            return ref == plan.image  # Agent present; sidecar missing.
-
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", side_effect=image_exists,
-        ), patch.object(
-            launch, "die", side_effect=Die(),
-        ):
-            with self.assertRaises(Die):
-                launch._build_images(plan)
-
-    def test_cached_mode_both_present_returns_unchanged_plan(self) -> None:
-        plan = self._cached_plan()
-        with patch.object(
-            launch, "read_committed_image", return_value="",
-        ), patch.object(
-            launch.container_mod, "image_exists", return_value=True,
-        ), patch.object(
-            launch.container_mod, "build_image",
-        ) as build, patch.object(launch, "info"):
-            result = launch._build_images(plan)
-
-        build.assert_not_called()
-        self.assertEqual(plan.image, result.image)
-
-    def test_committed_image_plus_cached_skips_sidecar_build(self) -> None:
-        plan = self._cached_plan()
-        with patch.object(
-            launch, "read_committed_image",
-            return_value="bot-bottle-committed-dev-abc:latest",
-        ), patch.object(
-            launch.container_mod, "image_exists", return_value=True,
-        ), patch.object(
-            launch.container_mod, "build_image",
-        ) as build, patch.object(launch, "info"):
-            result = launch._build_images(plan)
-
-        # In cached mode with a committed image, no builds should fire.
-        build.assert_not_called()
-        self.assertEqual("bot-bottle-committed-dev-abc:latest", result.image)
-
-
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_macos_container_util.py

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

tests/unit/test_smolmachines_launch_image.py

@@ -187,54 +187,6 @@ class TestAgentFromPath(unittest.TestCase):
             dockerfile="/repo/Dockerfile",
         )
 
-    def test_cached_policy_uses_existing_artifact_without_build(self):
-        with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
-            digest = "abcdef0123456789"
-            artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
-            artifact.write_text("")
-            plan = SimpleNamespace(
-                slug="dev-abc12",
-                agent_image="bot-bottle-claude:latest",
-                spec=SimpleNamespace(image_policy="cached"),
-            )
-            with patch.object(
-                _launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
-            ), patch.object(
-                _launch_mod, "read_committed_image", return_value="",
-            ), patch.object(
-                _launch_mod.docker_mod, "image_exists", return_value=True,
-            ), patch.object(
-                _launch_mod.docker_mod, "image_id",
-                return_value=f"sha256:{digest}fffffffffffffffff",
-            ), patch.object(
-                _launch_mod, "_ensure_smolmachine",
-            ) as ensure:
-                result = _launch_mod._agent_from_path(cast(Any, plan))
-
-        self.assertEqual(artifact, result)
-        ensure.assert_not_called()
-
-    def test_cached_policy_dies_when_artifact_missing(self):
-        with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
-            plan = SimpleNamespace(
-                slug="dev-abc12",
-                agent_image="bot-bottle-claude:latest",
-                spec=SimpleNamespace(image_policy="cached"),
-            )
-            with patch.object(
-                _launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
-            ), patch.object(
-                _launch_mod, "read_committed_image", return_value="",
-            ), patch.object(
-                _launch_mod.docker_mod, "image_exists", return_value=True,
-            ), patch.object(
-                _launch_mod.docker_mod, "image_id",
-                return_value="sha256:abcdef0123456789fffffffffffffffff",
-            ):
-                from bot_bottle.log import Die
-                with self.assertRaises(Die):
-                    _launch_mod._agent_from_path(cast(Any, plan))
-
 
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_sidecar_bundle.py

@@ -1,9 +1,4 @@
-"""Unit: bundle bringup primitives for the smolmachines backend
+"""Unit: bundle bringup primitives for the smolmachines backend."""
-(PRD 0023 chunk 2c).
-
-Tests mock `subprocess.run` and assert on the docker argv shape.
-The end-to-end integration smoke (real docker daemon, real
-bundle image) lands in chunk 2d."""
 
 from __future__ import annotations
 
@@ -17,28 +12,12 @@ from bot_bottle.backend.smolmachines.sidecar_bundle import (
     allocate_raw_host_ports,
     bundle_container_name,
     bundle_network_name,
-    create_bundle_network,
     ensure_bundle_image,
     start_bundle_vm,
-    remove_bundle_network,
-    start_bundle,
-    stop_bundle,
     stop_bundle_vm,
 )
 
 
-def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:  # type: ignore
-    return subprocess.CompletedProcess(
-        args=[], returncode=0, stdout=stdout, stderr=stderr,
-    )
-
-
-def _fail(stderr: str = "boom") -> subprocess.CompletedProcess:  # type: ignore
-    return subprocess.CompletedProcess(
-        args=[], returncode=1, stdout="", stderr=stderr,
-    )
-
-
 def _spec(**kwargs) -> BundleLaunchSpec:  # type: ignore
     defaults = dict(
         slug="demo-abc12",
@@ -71,123 +50,6 @@ class TestNamingHelpers(unittest.TestCase):
         )
 
 
-class TestNetworkLifecycle(unittest.TestCase):
-    def _patch_run(self, **kwargs):  # type: ignore
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            **kwargs,
-        )
-
-    def test_create_argv_explicit_subnet_and_gateway(self):
-        with self._patch_run(return_value=_ok()) as m:
-            create_bundle_network("nn", "192.168.50.0/24", "192.168.50.1")
-        self.assertEqual(
-            ["docker", "network", "create",
-             "--subnet", "192.168.50.0/24",
-             "--gateway", "192.168.50.1",
-             "nn"],
-            m.call_args.args[0],
-        )
-
-    def test_create_treats_existing_network_as_success(self):
-        with self._patch_run(return_value=_fail("network nn already exists")):
-            # No SystemExit.
-            create_bundle_network("nn", "192.168.50.0/24", "192.168.50.1")
-
-    def test_create_other_failure_is_fatal(self):
-        with self._patch_run(return_value=_fail("invalid subnet")):
-            with self.assertRaises(SystemExit):
-                create_bundle_network("nn", "bogus", "bogus")
-
-    def test_remove_missing_network_is_idempotent(self):
-        # No SystemExit / no warn-and-continue noise; missing
-        # network is the expected case during a partial teardown.
-        with self._patch_run(return_value=_fail("Error: No such network: nn")):
-            remove_bundle_network("nn")
-
-    def test_remove_clean_returns_success(self):
-        with self._patch_run(return_value=_ok()):
-            remove_bundle_network("nn")
-
-
-class TestStartBundle(unittest.TestCase):
-    def _patch_run(self):
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_ok(),
-        )
-
-    def test_argv_pins_ip_on_network(self):
-        with self._patch_run() as m:
-            start_bundle(_spec())
-        argv = m.call_args.args[0]
-        # --network NETNAME --ip <bundle-ip> on the docker run.
-        self.assertIn("--network", argv)
-        self.assertIn("bot-bottle-bundle-demo-abc12", argv)
-        self.assertIn("--ip", argv)
-        self.assertIn("192.168.50.2", argv)
-        # Detached and auto-removed.
-        self.assertIn("--detach", argv)
-        self.assertIn("--rm", argv)
-        # Container name uses the per-slug bundle prefix.
-        i = argv.index("--name")
-        self.assertEqual("bot-bottle-sidecars-demo-abc12", argv[i + 1])
-        # Image at the end.
-        self.assertEqual("bot-bottle-sidecars:latest", argv[-1])
-
-    def test_daemons_env_passed_in(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(daemons_csv="egress,supervise"))
-        argv = m.call_args.args[0]
-        self.assertIn("-e", argv)
-        self.assertIn(
-            "BOT_BOTTLE_SIDECAR_DAEMONS=egress,supervise",
-            argv,
-        )
-
-    def test_environment_entries_pass_through(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(environment=(
-                "SUPERVISE_BOTTLE_SLUG=demo-abc12",
-                "EGRESS_TOKEN_0",  # bare-name → host env inherit
-            )))
-        argv = m.call_args.args[0]
-        self.assertIn("SUPERVISE_BOTTLE_SLUG=demo-abc12", argv)
-        self.assertIn("EGRESS_TOKEN_0", argv)
-
-    def test_volumes_render_with_ro_flag(self):
-        with self._patch_run() as m:
-            start_bundle(_spec(volumes=(
-                ("/host/egress-ca.pem", "/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem", True),
-                ("/host/queue", "/run/supervise/queue", False),
-            )))
-        argv = m.call_args.args[0]
-        self.assertIn(
-            "/host/egress-ca.pem:/home/mitmproxy/.mitmproxy/mitmproxy-ca.pem:ro",
-            argv,
-        )
-        self.assertIn("/host/queue:/run/supervise/queue", argv)
-
-    def test_failure_dies(self):
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_fail("invalid mount"),
-        ):
-            with self.assertRaises(SystemExit):
-                start_bundle(_spec())
-
-    def test_host_env_inherited_to_subprocess(self):
-        # Bare-name entries in spec.environment rely on the docker
-        # subprocess being run with the host env. Confirm `env=`
-        # threads through.
-        with patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            return_value=_ok(),
-        ) as m:
-            start_bundle(_spec(), env={"FOO": "bar"})
-        self.assertEqual({"FOO": "bar"}, m.call_args.kwargs["env"])
-
-
 class TestEnsureBundleImage(unittest.TestCase):
     def test_builds_sidecar_dockerfile_before_plain_docker_run(self):
         with patch(
@@ -263,28 +125,6 @@ class TestStartBundleVm(unittest.TestCase):
         )
 
 
-class TestStopBundle(unittest.TestCase):
-    def _patch_run(self, **kwargs):  # type: ignore
-        return patch(
-            "bot_bottle.backend.smolmachines.sidecar_bundle.subprocess.run",
-            **kwargs,
-        )
-
-    def test_argv_force_removes(self):
-        with self._patch_run(return_value=_ok()) as m:
-            stop_bundle("demo-abc12")
-        self.assertEqual(
-            ["docker", "rm", "-f", "bot-bottle-sidecars-demo-abc12"],
-            m.call_args.args[0],
-        )
-
-    def test_missing_container_is_idempotent(self):
-        with self._patch_run(return_value=_fail(
-            "Error: No such container: bot-bottle-sidecars-demo-abc12"
-        )):
-            stop_bundle("demo-abc12")  # no raise
-
-
 class TestStopBundleVm(unittest.TestCase):
     def test_stops_then_deletes_sidecar_vm(self):
         with patch(

tests/unit/test_stale_checks.py

@@ -1,302 +0,0 @@
-"""Unit: stale-image check functions across backends, and the
-BottleBackend.launch template method (skip_stale flag).
-
-No real images or containers are used — all Docker/container/smolmachine
-calls are mocked at the module boundary."""
-
-from __future__ import annotations
-
-import tempfile
-import unittest
-from pathlib import Path
-from types import SimpleNamespace
-from typing import Any, cast
-from unittest.mock import MagicMock, patch
-
-
-def _bottle_cm(bottle: Any) -> MagicMock:
-    """Return a mock context manager that yields `bottle`."""
-    cm = MagicMock()
-    cm.__enter__ = MagicMock(return_value=bottle)
-    cm.__exit__ = MagicMock(return_value=False)
-    return cm
-
-
-# ---------------------------------------------------------------------------
-# BottleBackend.launch template — _image_stale_checks + skip_stale
-# ---------------------------------------------------------------------------
-
-class TestBottleBackendLaunchTemplate(unittest.TestCase):
-    """Verify the concrete launch() method on BottleBackend calls
-    _image_stale_checks unless skip_stale=True, then delegates to _launch_impl."""
-
-    def _make_backend(self) -> Any:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        return DockerBottleBackend()
-
-    def test_stale_checks_called_by_default(self) -> None:
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ):
-            with backend.launch(plan):
-                pass
-        stale_mock.assert_called_once_with(plan)
-
-    def test_skip_stale_bypasses_stale_checks(self) -> None:
-        backend = self._make_backend()
-        plan = cast(Any, SimpleNamespace())
-        bottle = MagicMock()
-        with patch.object(
-            backend, "_image_stale_checks",
-        ) as stale_mock, patch.object(
-            backend, "_launch_impl",
-            return_value=_bottle_cm(bottle),
-        ):
-            with backend.launch(plan, skip_stale=True):
-                pass
-        stale_mock.assert_not_called()
-
-    def test_noop_default_image_stale_checks(self) -> None:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        from bot_bottle.backend import BottleBackend
-        backend = DockerBottleBackend()
-        # Call the base-class _image_stale_checks directly to verify it's a no-op.
-        BottleBackend._image_stale_checks(backend, cast(Any, SimpleNamespace()))  # type: ignore[arg-type]
-
-
-# ---------------------------------------------------------------------------
-# Docker backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestDockerStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
-        spec = SimpleNamespace(image_policy=policy)
-        provision = SimpleNamespace(image="bot-bottle-agent:latest")
-        return cast(Any, SimpleNamespace(
-            spec=spec,
-            slug=slug,
-            image="bot-bottle-agent:latest",
-            agent_provision=provision,
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        cs.assert_not_called()
-
-    def test_committed_image_present_checks_only_committed(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
-             patch.object(mod.docker_mod, "image_exists", return_value=True), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_called_once()
-        self.assertIn("committed:latest", cs.call_args.args[0])
-
-    def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", return_value=True), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        # Should have been called for both agent and sidecar images.
-        self.assertEqual(2, cs.call_count)
-
-    def test_image_not_present_skips_check(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", return_value=False), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_not_called()
-
-    def test_only_sidecar_missing_checks_only_agent(self) -> None:
-        from bot_bottle.backend.docker import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        plan = self._plan()
-
-        def image_exists(ref: str) -> bool:
-            return ref == plan.image  # Only agent present; sidecar missing.
-
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \
-             patch.object(mod.docker_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(plan)
-        self.assertEqual(1, cs.call_count)
-        self.assertIn(plan.image, cs.call_args.args[0])
-
-    def test_backend_image_stale_checks_delegates(self) -> None:
-        from bot_bottle.backend.docker.backend import DockerBottleBackend
-        from bot_bottle.backend.docker import launch as mod
-        backend = DockerBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# smolmachines backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestSmolmachinesStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached") -> Any:
-        return cast(Any, SimpleNamespace(
-            spec=SimpleNamespace(image_policy=policy),
-            slug="dev-abc",
-            agent_image="bot-bottle-claude:latest",
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale_path") as csp:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        csp.assert_not_called()
-
-    def test_committed_artifact_found_checks_it(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        with tempfile.NamedTemporaryFile(suffix=".smolmachine") as f:
-            committed_path = f.name
-            with patch.object(mod, "read_committed_image", return_value=committed_path), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=False), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        csp.assert_called_once()
-        self.assertIn("agent smolmachine", csp.call_args.args[0])
-
-    def test_no_committed_agent_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        digest = "abcdef0123456789"
-        with tempfile.TemporaryDirectory() as tmp:
-            artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
-            artifact.write_text("")
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 patch.object(mod, "read_committed_image", return_value=""), \
-                 patch.object(mod.docker_mod, "image_exists", return_value=True), \
-                 patch.object(
-                     mod.docker_mod, "image_id",
-                     return_value=f"sha256:{digest}ffffffffffffffff",
-                 ), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        # At least the agent artifact check ran.
-        self.assertGreaterEqual(csp.call_count, 1)
-
-    def test_sidecar_artifact_in_cache_is_checked(self) -> None:
-        from bot_bottle.backend.smolmachines import launch as mod
-        sidecar_digest = "fedcba9876543210"
-        with tempfile.TemporaryDirectory() as tmp:
-            sidecar_artifact = (
-                Path(tmp) / f"{sidecar_digest}.smolmachine.smolmachine"
-            )
-            sidecar_artifact.write_text("")
-
-            def image_exists(ref: str) -> bool:  # pylint: disable=unused-argument
-                return True
-
-            def image_id(ref: str) -> str:
-                if ref == mod._bundle.SIDECAR_BUNDLE_IMAGE:  # type: ignore[attr-defined]
-                    return f"sha256:{sidecar_digest}ffffffff"
-                return "sha256:0000000000000000ffffffff"
-
-            with patch.object(mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp)), \
-                 patch.object(mod, "read_committed_image", return_value=""), \
-                 patch.object(mod.docker_mod, "image_exists", side_effect=image_exists), \
-                 patch.object(mod.docker_mod, "image_id", side_effect=image_id), \
-                 patch.object(mod, "check_stale_path") as csp:
-                mod.stale_checks(self._plan())
-        sidecar_calls = [c for c in csp.call_args_list if "sidecar" in c.args[0]]
-        self.assertGreaterEqual(len(sidecar_calls), 1)
-
-    def test_backend_image_stale_checks_delegates(self) -> None:
-        from bot_bottle.backend.smolmachines.backend import SmolmachinesBottleBackend
-        from bot_bottle.backend.smolmachines import launch as mod
-        backend = SmolmachinesBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-# ---------------------------------------------------------------------------
-# macOS container backend stale_checks
-# ---------------------------------------------------------------------------
-
-class TestMacosContainerStaleChecks(unittest.TestCase):
-    def _plan(self, policy: str = "cached", slug: str = "dev-abc") -> Any:
-        return cast(Any, SimpleNamespace(
-            spec=SimpleNamespace(image_policy=policy),
-            slug=slug,
-            image="bot-bottle-agent:latest",
-        ))
-
-    def test_fresh_policy_is_noop(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        with patch.object(mod, "read_committed_image") as rci, \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan("fresh"))
-        rci.assert_not_called()
-        cs.assert_not_called()
-
-    def test_committed_image_present_checks_only_committed(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value="committed:latest"), \
-             patch.object(mod.container_mod, "image_exists", return_value=True), \
-             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_called_once()
-        self.assertIn("committed:latest", cs.call_args.args[0])
-
-    def test_no_committed_image_checks_agent_and_sidecar(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        from datetime import datetime, timezone
-        ts = datetime(2025, 1, 1, tzinfo=timezone.utc)
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.container_mod, "image_exists", return_value=True), \
-             patch.object(mod.container_mod, "image_created_at", return_value=ts), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        self.assertEqual(2, cs.call_count)
-
-    def test_image_not_present_skips_check(self) -> None:
-        from bot_bottle.backend.macos_container import launch as mod
-        with patch.object(mod, "read_committed_image", return_value=""), \
-             patch.object(mod.container_mod, "image_exists", return_value=False), \
-             patch.object(mod, "check_stale") as cs:
-            mod.stale_checks(self._plan())
-        cs.assert_not_called()
-
-    def test_backend_image_stale_checks_delegates(self) -> None:
-        from bot_bottle.backend.macos_container.backend import MacosContainerBottleBackend
-        from bot_bottle.backend.macos_container import launch as mod
-        backend = MacosContainerBottleBackend()
-        plan = self._plan()
-        with patch.object(mod, "stale_checks") as sc:
-            backend._image_stale_checks(plan)  # type: ignore[arg-type]
-        sc.assert_called_once_with(plan)
-
-
-if __name__ == "__main__":
-    unittest.main()