fix(tests): resolve pyright errors in stale-check test files

Remove unused imports, add missing type annotations, fix Die()
constructor calls (int code, not str), replace fake backend
subclass with patch.object approach to avoid reportMissingParameterType
and override-incompatibility errors in strict pyright mode.

bot_bottle/backend/__init__.py

@@ -36,10 +36,10 @@ import os
 import shlex
 import sys
 from abc import ABC, abstractmethod
-from contextlib import AbstractContextManager
+from contextlib import AbstractContextManager, contextmanager
 from dataclasses import dataclass
 from pathlib import Path
-from typing import Any, Generic, Sequence, TypeVar
+from typing import Any, Generator, Generic, Sequence, TypeVar
 
 from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
 from ..egress import EgressPlan
@@ -78,6 +78,9 @@ class BottleSpec:
     # True when launched via --headless (no TTY, no interactive prompts).
     # The git-gate host-key preflight uses this to error rather than prompt.
     headless: bool = False
+    # Image startup policy. "fresh" preserves the normal build path;
+    # "cached" reuses the current local image/artifact without rebuilding.
+    image_policy: str = "fresh"
 
 
 @dataclass(frozen=True)
@@ -432,8 +435,25 @@ 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(self, plan: PlanT) -> AbstractContextManager[Bottle]:
+    def _launch_impl(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,8 +85,11 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: DockerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
+    def _launch_impl(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,10 +180,6 @@ 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},
@@ -192,6 +188,11 @@ 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,7 +41,8 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale
+from ...log import die, info, warn
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -63,6 +64,7 @@ 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.
@@ -100,12 +102,26 @@ 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,
@@ -207,3 +223,21 @@ def launch(
         yield bottle
     finally:
         teardown()
+
+
+def stale_checks(plan: DockerBottlePlan) -> None:
+    """Raise StaleImageError if a cached image is older than the configured
+    threshold. Only runs when image_policy is 'cached'. Called by the backend
+    class's _image_stale_checks before _launch_impl starts any resources."""
+    if plan.spec.image_policy != "cached":
+        return
+    committed = read_committed_image(plan.slug)
+    if committed and docker_mod.image_exists(committed):
+        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,6 +4,7 @@ existence, and building images."""
 
 from __future__ import annotations
 
+from datetime import datetime, timezone
 import re
 import shutil
 import subprocess
@@ -186,6 +187,45 @@ 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,8 +67,11 @@ class MacosContainerBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: MacosContainerBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
+    def _launch_impl(
         self, plan: MacosContainerBottlePlan
     ) -> Generator[MacosContainerBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/macos_container/launch.py

@@ -32,6 +32,7 @@ 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
@@ -149,27 +150,51 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
 
 
 def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
-    container_mod.build_image(
-        SIDECAR_BUNDLE_IMAGE,
-        _REPO_DIR,
-        dockerfile=SIDECAR_BUNDLE_DOCKERFILE,
-    )
+    cached = plan.spec.image_policy == "cached"
     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),
         )
-    container_mod.build_image(
-        plan.image,
-        _REPO_DIR,
-        dockerfile=plan.dockerfile_path,
+    if cached:
+        if not container_mod.image_exists(plan.image):
+            die(
+                f"cached agent image {plan.image!r} not found; "
+                "run without --cached-images to build it"
             )
+        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(

bot_bottle/backend/macos_container/util.py

@@ -10,6 +10,7 @@ import shutil
 import subprocess
 import tempfile
 import time
+from datetime import datetime, timezone
 from typing import Iterable
 
 from ...log import die, info
@@ -458,6 +459,39 @@ 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,8 +77,11 @@ class SmolmachinesBottleBackend(
             stage_dir=stage_dir,
         )
 
+    def _image_stale_checks(self, plan: SmolmachinesBottlePlan) -> None:
+        _launch.stale_checks(plan)
+
     @contextmanager
-    def launch(
+    def _launch_impl(
         self, plan: SmolmachinesBottlePlan
     ) -> Generator[SmolmachinesBottle, None, None]:
         with _launch.launch(plan, provision=self.provision) as bottle:

bot_bottle/backend/smolmachines/egress_apply.py

@@ -1,20 +1,51 @@
 """Egress apply for the smolmachines backend.
 
-The smolmachines sidecar bundle runs as a host-side Docker container,
-so egress signalling is identical to the docker backend.
+The smolmachines sidecar bundle runs as a sidecar smolVM. Route-file
+inspection and reload signalling therefore go through ``smolvm machine
+exec`` instead of Docker.
 """
 
 from __future__ import annotations
 
-from ..docker.egress_apply import (  # noqa: F401
-    DockerEgressApplicator,
-    EgressApplyError,
-    applicator,
-    fetch_current_routes,
+from ...egress import EGRESS_ROUTES_IN_CONTAINER
+from ...log import warn
+from ..egress_apply import EgressApplicator, EgressApplyError
+from . import sidecar_bundle as _bundle
+from . import smolvm as _smolvm
+
+
+def fetch_current_routes(slug: str) -> str:
+    machine = _bundle.bundle_machine_name(slug)
+    result = _smolvm.machine_exec(machine, ["cat", EGRESS_ROUTES_IN_CONTAINER])
+    if result.returncode != 0:
+        raise EgressApplyError(
+            f"could not read routes.yaml from {machine}: "
+            f"{(result.stderr or '').strip() or 'sidecar VM not running?'}"
+        )
+    return result.stdout
+
+
+class SmolmachinesEgressApplicator(EgressApplicator):
+    def _signal_bundle_reload(self, slug: str) -> None:
+        machine = _bundle.bundle_machine_name(slug)
+        result = _smolvm.machine_exec(machine, ["sh", "-c", "kill -HUP 1"])
+        if result.returncode != 0:
+            last_error = (result.stderr or "").strip() or (result.stdout or "").strip()
+            warn(
+                f"egress: routes updated on disk for {slug}, but bundle reload failed: "
+                f"{last_error or 'smolvm exec failed'}"
+            )
+            raise EgressApplyError(
+                f"could not reload egress bundle {machine}: "
+                f"{last_error or 'smolvm exec failed'}"
             )
 
+
+applicator = SmolmachinesEgressApplicator()
+
+
 __all__ = [
-    "DockerEgressApplicator",
+    "SmolmachinesEgressApplicator",
     "EgressApplyError",
     "applicator",
     "fetch_current_routes",

bot_bottle/backend/smolmachines/enumerate.py

@@ -31,6 +31,7 @@ from . import sidecar_bundle as _bundle
 # matching the bundle container name pattern. We use the prefix
 # both as a filter and to strip back to the slug.
 _VM_NAME_PREFIX = "bot-bottle-"
+_SIDECAR_VM_PREFIX = _bundle.bundle_machine_name("")
 
 
 def enumerate_active() -> list[ActiveAgent]:
@@ -54,7 +55,11 @@ def enumerate_active() -> list[ActiveAgent]:
     for m in machines:
         name = m.get("name") or ""
         state = m.get("state") or ""
-        if state != "running" or not name.startswith(_VM_NAME_PREFIX):
+        if (
+            state != "running"
+            or not name.startswith(_VM_NAME_PREFIX)
+            or name.startswith(_SIDECAR_VM_PREFIX)
+        ):
             continue
         slug = name[len(_VM_NAME_PREFIX):]
         metadata = read_metadata(slug)

bot_bottle/backend/smolmachines/launch.py

@@ -17,7 +17,6 @@ from __future__ import annotations
 
 import dataclasses
 import os
-import platform
 from contextlib import ExitStack, contextmanager
 from pathlib import Path
 from typing import Callable, Generator
@@ -46,13 +45,15 @@ from ...git_gate import (
     provision_git_gate_dynamic_keys,
     revoke_git_gate_provisioned_keys,
 )
-from ...log import info, warn
+from ...image_cache import check_stale_path
+from ...log import die, info, warn
 from ...bottle_state import (
     egress_state_dir,
     git_gate_state_dir,
     read_committed_image,
 )
 from . import loopback_alias as _loopback
+from . import port_forward as _forward
 from . import sidecar_bundle as _bundle
 from . import smolvm as _smolvm
 from .bottle import SmolmachinesBottle
@@ -92,9 +93,8 @@ def launch(
     try:
         loopback_ip, network = _allocate_resources(plan, stack)
         plan = _mint_certs(plan)
-        proxy_host = _proxy_host(plan, loopback_ip)
+        proxy_host = loopback_ip
         plan = _start_bundle(plan, network, proxy_host, stack)
-        plan = _discover_urls(plan, proxy_host)
 
         agent_from_path = _agent_from_path(plan)
 
@@ -144,19 +144,17 @@ def _allocate_resources(
     plan: SmolmachinesBottlePlan,
     stack: ExitStack,
 ) -> tuple[str, str]:
-    """Reserve a loopback alias and create the per-bottle docker bridge.
+    """Reserve a per-bottle host address.
 
-    The per-bottle alias scopes TSI's allowlist to this bottle's
-    published ports so the agent can't reach other bottles' or host
-    services' ports on loopback. On macOS `ensure_pool` first
-    sudo-aliases the pool on `lo0`; on Linux that's a no-op since
-    all of 127.0.0.0/8 is already loopback, but the per-bottle
-    allocation runs on both."""
+    The per-bottle address scopes TSI's allowlist to this bottle's
+    forwarder-published ports so the agent can't reach other bottles'
+    or host services. The returned network name remains in the bundle
+    spec for compatibility with older helper tests; no Docker sidecar
+    container is launched."""
+    del stack
     _loopback.ensure_pool()
     loopback_ip = _loopback.allocate(plan.slug)
     network = _bundle.bundle_network_name(plan.slug)
-    _bundle.create_bundle_network(network, plan.bundle_subnet, plan.bundle_gateway)
-    stack.callback(_bundle.remove_bundle_network, network)
     return loopback_ip, network
 
 
@@ -179,15 +177,43 @@ def _start_bundle(
     proxy_host: str,
     stack: ExitStack,
 ) -> SmolmachinesBottlePlan:
-    """Build the BundleLaunchSpec, resolve token env, start the
-    sidecar bundle container, and register teardown."""
+    """Build the BundleLaunchSpec, start the sidecar VM, wrap its raw
+    smolVM-published loopback ports with per-bottle forwarders, stamp
+    agent URLs from those forwarder ports, and register teardown."""
     plan = _provision_git_gate_keys(plan)
     bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
     token_env = _resolve_token_env(plan, dict(os.environ))
-    _bundle.ensure_bundle_image(bundle_spec.image)
-    _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
-    stack.callback(_bundle.stop_bundle, plan.slug)
-    return plan
+    if _image_policy(plan) == "cached":
+        artifact = _cached_smolmachine(bundle_spec.image, label="sidecar")
+    else:
+        artifact = _ensure_smolmachine(
+            bundle_spec.image,
+            dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
+        )
+    launch = _bundle.start_bundle_vm(
+        bundle_spec,
+        from_path=artifact,
+        host_env={**os.environ, **token_env},
+    )
+    stack.callback(_bundle.stop_bundle_vm, plan.slug)
+
+    forward_specs = tuple(
+        _forward.ForwardSpec(
+            label=_label_for_port(container_port),
+            listen_host=proxy_host,
+            listen_port=0,
+            target_host="127.0.0.1",
+            target_port=host_port,
+        )
+        for container_port, host_port in launch.raw_ports.items()
+    )
+    handle = _forward.start_forwarder(forward_specs)
+    stack.callback(_forward.stop_forwarder, handle)
+    published_ports = {
+        _port_for_label(spec.label): spec.listen_port
+        for spec in handle.forwards
+    }
+    return _discover_urls(plan, proxy_host, published_ports)
 
 
 def _provision_git_gate_keys(
@@ -206,35 +232,27 @@ def _provision_git_gate_keys(
 def _discover_urls(
     plan: SmolmachinesBottlePlan,
     proxy_host: str,
+    published_ports: dict[int, int],
 ) -> SmolmachinesBottlePlan:
-    """Discover host-side ports for published container ports and
-    return the plan with URLs + guest_env stamped in.
+    """Stamp URLs + guest_env from per-bottle forwarder ports.
 
-    `proxy_host` is the host IP that both TSI's allowlist and
-    docker's port-forward bindings are keyed to. On macOS it is the
-    per-bottle loopback alias; on Linux it is the per-bottle bridge
-    gateway (see `_proxy_host`). The agent dials the published port
-    on this IP for all bundle-hosted services.
+    `proxy_host` is the host IP that both TSI's allowlist and the
+    forwarder listeners are keyed to. The raw smolVM-published ports
+    are intentionally not advertised to the agent.
 
     NO_PROXY includes `proxy_host` so supervise + git-gate URLs
     bypass HTTPS_PROXY."""
-    agent_facing_host_port = _bundle.bundle_host_port(
-        plan.slug, _EGRESS_PORT, host_ip=proxy_host,
-    )
+    agent_facing_host_port = published_ports[_EGRESS_PORT]
     agent_proxy_url = f"http://{proxy_host}:{agent_facing_host_port}"
 
     agent_git_gate_host = ""
     if plan.git_gate_plan.upstreams:
-        git_gate_host_port = _bundle.bundle_host_port(
-            plan.slug, _GIT_HTTP_PORT, host_ip=proxy_host,
-        )
+        git_gate_host_port = published_ports[_GIT_HTTP_PORT]
         agent_git_gate_host = f"{proxy_host}:{git_gate_host_port}"
 
     agent_supervise_url = ""
     if plan.supervise_plan is not None:
-        supervise_host_port = _bundle.bundle_host_port(
-            plan.slug, _SUPERVISE_PORT, host_ip=proxy_host,
-        )
+        supervise_host_port = published_ports[_SUPERVISE_PORT]
         agent_supervise_url = f"http://{proxy_host}:{supervise_host_port}/"
 
     existing_no_proxy = plan.guest_env.get("NO_PROXY", "localhost,127.0.0.1")
@@ -273,14 +291,13 @@ def _launch_vm(
 ) -> None:
     """Create, patch, and start the smolvm VM; register teardown.
 
-    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias
-    on macOS or the bridge gateway on Linux (see `_proxy_host`). This
-    ensures the guest can only reach bundle ports published on that IP,
-    not the container IP directly. force_allowlist confirms the
-    allowlist persisted (patching smolvm 0.8.0's silent-drop of
-    --allow-cidr when combined with --from) and fails closed if it
-    can't. Smolfile isn't usable here — smolvm 0.8.0 makes --from
-    and --smolfile mutually exclusive."""
+    --allow-cidr is `proxy_host/32` — the per-bottle loopback alias.
+    This ensures the guest can only reach sidecar forwarders published
+    on that IP, not host localhost or another bottle's alias.
+    force_allowlist confirms the allowlist persisted (patching smolvm
+    0.8.0's silent-drop of --allow-cidr when combined with --from) and
+    fails closed if it can't. Smolfile isn't usable here — smolvm 0.8.0
+    makes --from and --smolfile mutually exclusive."""
     tsi_cidr = f"{proxy_host}/32"
     _smolvm.machine_create(
         plan.machine_name,
@@ -327,25 +344,26 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
     _smolvm.wait_exec_ready(plan.machine_name)
 
 
-def _proxy_host(plan: SmolmachinesBottlePlan, loopback_ip: str) -> str:
-    """Return the host IP for TSI's allowlist and docker port-forward bindings.
+def _label_for_port(port: int) -> str:
+    if port == _EGRESS_PORT:
+        return "egress"
+    if port == _GIT_HTTP_PORT:
+        return "git-http"
+    if port == _SUPERVISE_PORT:
+        return "supervise"
+    return f"port-{port}"
 
-    On macOS, the per-bottle loopback alias (e.g. ``127.0.0.16``) works
-    because macOS's network stack lets TSI intercept 127.x.x.x connects
-    from the guest before they reach the host's own loopback.
 
-    On Linux, the guest kernel's LOCAL routing table routes all
-    ``127.0.0.0/8`` to the guest's own loopback — those packets never
-    reach eth0 and TSI never sees them. Using the per-bottle bridge
-    gateway (e.g. ``192.168.N.1``) instead sidesteps the problem: it
-    is not a loopback address, so the guest routes it via eth0 and TSI
-    intercepts it normally. The TSI allowlist is ``gateway/32``, which
-    is distinct from the container IP (``192.168.N.2``), so the agent
-    still can't reach the egress daemon directly — TSI blocks any
-    connection to the container IP that isn't via the published port."""
-    if platform.system() == "Linux":
-        return plan.bundle_gateway
-    return loopback_ip
+def _port_for_label(label: str) -> int:
+    if label == "egress":
+        return _EGRESS_PORT
+    if label == "git-http":
+        return _GIT_HTTP_PORT
+    if label == "supervise":
+        return _SUPERVISE_PORT
+    if label.startswith("port-"):
+        return int(label.removeprefix("port-"))
+    raise ValueError(f"unknown sidecar forward label: {label}")
 
 
 def _bundle_launch_spec(
@@ -455,6 +473,9 @@ 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
@@ -465,6 +486,60 @@ 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/port_forward.py

@@ -0,0 +1,245 @@
+"""Per-bottle TCP forwarder for smolmachines sidecar VMs.
+
+smolVM currently publishes guest ports on host loopback as
+``HOST:GUEST`` port pairs, without an address-scoped bind. The agent
+VM's TSI allowlist is IP-only, so bot-bottle exposes a second, stricter
+surface: listeners bound only to the bottle's allocated host address,
+forwarding to the raw smolVM-published loopback ports.
+"""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import selectors
+import signal
+import socket
+import subprocess
+import sys
+import threading
+from dataclasses import asdict, dataclass
+from typing import Any, Iterable, Sequence
+
+from ...log import die, warn
+
+
+_BUFFER_SIZE = 64 * 1024
+_LOOPBACK_TARGETS = {"127.0.0.1", "::1"}
+_FORBIDDEN_LISTEN_HOSTS = {"", "0.0.0.0", "::", "127.0.0.1", "::1"}
+
+
+@dataclass(frozen=True)
+class ForwardSpec:
+    label: str
+    listen_host: str
+    listen_port: int
+    target_host: str
+    target_port: int
+
+
+@dataclass(frozen=True)
+class ForwarderHandle:
+    process: Any
+    forwards: tuple[ForwardSpec, ...]
+
+
+def start_forwarder(specs: Sequence[ForwardSpec]) -> ForwarderHandle:
+    """Start one helper process for a bottle and wait until all
+    listeners are bound. ``listen_port`` may be 0; the returned specs
+    contain the kernel-assigned ports printed by the helper."""
+    if not specs:
+        return ForwarderHandle(process=_CompletedProcess(), forwards=())
+    _validate_specs(specs)
+    argv = [
+        sys.executable,
+        "-c",
+        "from bot_bottle.backend.smolmachines.port_forward import main; "
+        "raise SystemExit(main())",
+        "--spec-json",
+        json.dumps([asdict(s) for s in specs], separators=(",", ":")),
+    ]
+    proc = subprocess.Popen(
+        argv,
+        stdin=subprocess.DEVNULL,
+        stdout=subprocess.PIPE,
+        stderr=None,
+        text=True,
+        env=os.environ,
+    )
+    assert proc.stdout is not None
+    line = proc.stdout.readline()
+    proc.stdout.close()
+    if not line:
+        proc.terminate()
+        proc.wait(timeout=5)
+        die("smolmachines forwarder failed before reporting readiness")
+    try:
+        payload = json.loads(line)
+        bound = tuple(ForwardSpec(**item) for item in payload["forwards"])
+    except (KeyError, TypeError, ValueError, json.JSONDecodeError) as exc:
+        proc.terminate()
+        proc.wait(timeout=5)
+        die(f"smolmachines forwarder returned invalid readiness payload: {exc}")
+    return ForwarderHandle(process=proc, forwards=bound)
+
+
+def stop_forwarder(handle: ForwarderHandle) -> None:
+    proc = handle.process
+    if proc.poll() is not None:
+        return
+    proc.terminate()
+    try:
+        proc.wait(timeout=5)
+    except subprocess.TimeoutExpired:
+        proc.kill()
+        proc.wait(timeout=5)
+
+
+def _validate_specs(specs: Iterable[ForwardSpec]) -> None:
+    for spec in specs:
+        if spec.listen_host in _FORBIDDEN_LISTEN_HOSTS:
+            die(f"refusing unsafe smolmachines forwarder bind: {spec.listen_host!r}")
+        if spec.target_host not in _LOOPBACK_TARGETS:
+            die(f"refusing non-loopback smolmachines forwarder target: {spec.target_host!r}")
+        if not 0 <= spec.listen_port <= 65535:
+            die(f"invalid smolmachines forwarder listen port: {spec.listen_port}")
+        if not 1 <= spec.target_port <= 65535:
+            die(f"invalid smolmachines forwarder target port: {spec.target_port}")
+
+
+class _CompletedProcess:
+    """Popen-like no-op used when a bottle has no forwards."""
+
+    def poll(self) -> int:
+        return 0
+
+    def terminate(self) -> None:
+        return None
+
+    def kill(self) -> None:
+        return None
+
+    def wait(self, timeout: float | None = None) -> int:
+        del timeout
+        return 0
+
+
+class _Forwarder:
+    def __init__(self, specs: Sequence[ForwardSpec]):
+        _validate_specs(specs)
+        self.specs = tuple(specs)
+        self.selector = selectors.DefaultSelector()
+        self.stop_event = threading.Event()
+        self.listeners: list[tuple[socket.socket, ForwardSpec]] = []
+        self.bound: list[ForwardSpec] = []
+
+    def start(self) -> None:
+        for spec in self.specs:
+            listener = socket.create_server(
+                (spec.listen_host, spec.listen_port),
+                reuse_port=False,
+                backlog=64,
+            )
+            listener.setblocking(False)
+            host, port = listener.getsockname()[:2]
+            bound = ForwardSpec(
+                label=spec.label,
+                listen_host=str(host),
+                listen_port=int(port),
+                target_host=spec.target_host,
+                target_port=spec.target_port,
+            )
+            self.listeners.append((listener, bound))
+            self.bound.append(bound)
+            self.selector.register(listener, selectors.EVENT_READ, bound)
+
+    def serve(self) -> None:
+        while not self.stop_event.is_set():
+            try:
+                events = self.selector.select(timeout=0.25)
+            except OSError:
+                return
+            for key, _ in events:
+                listener = key.fileobj
+                if not isinstance(listener, socket.socket):
+                    continue
+                spec = key.data
+                try:
+                    client, _ = listener.accept()
+                except OSError:
+                    continue
+                threading.Thread(
+                    target=self._handle_client,
+                    args=(client, spec),
+                    daemon=True,
+                ).start()
+
+    def stop(self) -> None:
+        self.stop_event.set()
+        for listener, _ in self.listeners:
+            try:
+                self.selector.unregister(listener)
+            except Exception:  # noqa: BLE001 - best effort shutdown
+                pass
+            listener.close()
+        self.selector.close()
+
+    def _handle_client(self, client: socket.socket, spec: ForwardSpec) -> None:
+        with client:
+            try:
+                target = socket.create_connection((spec.target_host, spec.target_port))
+            except OSError as exc:
+                warn(f"smolmachines forwarder {spec.label}: target connect failed: {exc}")
+                return
+            with target:
+                left = threading.Thread(target=_pipe, args=(client, target), daemon=True)
+                right = threading.Thread(target=_pipe, args=(target, client), daemon=True)
+                left.start()
+                right.start()
+                left.join()
+                right.join()
+
+
+def _pipe(src: socket.socket, dst: socket.socket) -> None:
+    while True:
+        try:
+            data = src.recv(_BUFFER_SIZE)
+        except OSError:
+            break
+        if not data:
+            break
+        try:
+            dst.sendall(data)
+        except OSError:
+            break
+    try:
+        dst.shutdown(socket.SHUT_WR)
+    except OSError:
+        pass
+
+
+def main(argv: Sequence[str] | None = None) -> int:
+    parser = argparse.ArgumentParser()
+    parser.add_argument("--spec-json", required=True)
+    ns = parser.parse_args(argv)
+    specs = tuple(ForwardSpec(**item) for item in json.loads(ns.spec_json))
+    forwarder = _Forwarder(specs)
+
+    def request_stop(signum: int, _frame: object) -> None:
+        del signum
+        forwarder.stop()
+
+    signal.signal(signal.SIGTERM, request_stop)
+    signal.signal(signal.SIGINT, request_stop)
+    forwarder.start()
+    print(json.dumps({"forwards": [asdict(s) for s in forwarder.bound]}), flush=True)
+    try:
+        forwarder.serve()
+    finally:
+        forwarder.stop()
+    return 0
+
+
+if __name__ == "__main__":
+    raise SystemExit(main())

bot_bottle/backend/smolmachines/sidecar_bundle.py

@@ -23,6 +23,8 @@ 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
@@ -34,6 +36,7 @@ from ..docker.sidecar_bundle import (
     SIDECAR_BUNDLE_DOCKERFILE,
     SIDECAR_BUNDLE_IMAGE,
 )
+from . import smolvm as _smolvm
 
 
 _REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
@@ -54,6 +57,11 @@ def bundle_container_name(slug: str) -> str:
     return f"bot-bottle-sidecars-{slug}"
 
 
+def bundle_machine_name(slug: str) -> str:
+    """Sidecar smolVM machine name."""
+    return bundle_container_name(slug)
+
+
 @dataclass(frozen=True)
 class BundleLaunchSpec:
     """Everything `start_bundle` needs to bring up one bundle
@@ -92,6 +100,11 @@ class BundleLaunchSpec:
     publish_host_ip: str = "127.0.0.1"
 
 
+@dataclass(frozen=True)
+class BundleVmLaunch:
+    raw_ports: dict[int, int]
+
+
 def ensure_bundle_image(image: str = SIDECAR_BUNDLE_IMAGE) -> None:
     """Build the sidecar bundle image before `docker run`.
 
@@ -107,6 +120,70 @@ def ensure_bundle_image(image: str = SIDECAR_BUNDLE_IMAGE) -> None:
     )
 
 
+def allocate_raw_host_ports(container_ports: Sequence[int]) -> dict[int, int]:
+    """Reserve candidate host loopback ports for smolVM `-p HOST:GUEST`.
+
+    The sockets are closed before smolVM binds them, so this remains
+    best-effort. smolVM failing to bind a selected port is fatal in
+    the caller's launch path."""
+    out: dict[int, int] = {}
+    for container_port in container_ports:
+        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as sock:
+            sock.bind(("127.0.0.1", 0))
+            out[container_port] = int(sock.getsockname()[1])
+    return out
+
+
+def start_bundle_vm(
+    spec: BundleLaunchSpec,
+    *,
+    from_path: Path,
+    host_env: dict[str, str] | None = None,
+) -> BundleVmLaunch:
+    """Create and start the sidecar bundle as a smolVM.
+
+    smolVM's own published ports are raw host-loopback ports. The
+    launch flow wraps them with per-bottle address-bound forwarders
+    before exposing anything to the agent VM."""
+    raw_ports = allocate_raw_host_ports(spec.ports_to_publish)
+    effective_host_env = host_env if host_env is not None else os.environ
+    env: dict[str, str] = {"BOT_BOTTLE_SIDECAR_DAEMONS": spec.daemons_csv}
+    for entry in spec.environment:
+        name, sep, value = entry.partition("=")
+        if sep:
+            env[name] = value
+        elif entry in effective_host_env:
+            env[entry] = effective_host_env[entry]
+    name = bundle_machine_name(spec.slug)
+    _smolvm.machine_create(
+        name,
+        from_path=from_path,
+        net=True,
+        env=env,
+        volumes=spec.volumes,
+        ports=tuple(
+            (host_port, guest_port)
+            for guest_port, host_port in raw_ports.items()
+        ),
+    )
+    _smolvm.machine_start(name)
+    _smolvm.wait_exec_ready(name)
+    return BundleVmLaunch(raw_ports=raw_ports)
+
+
+def stop_bundle_vm(slug: str) -> None:
+    """Best-effort sidecar VM teardown."""
+    name = bundle_machine_name(slug)
+    try:
+        _smolvm.machine_stop(name)
+    except _smolvm.SmolvmError as exc:
+        warn(f"smolvm machine stop {name} failed: {exc}")
+    try:
+        _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

bot_bottle/backend/smolmachines/smolvm.py

@@ -113,8 +113,11 @@ def machine_create(
     *,
     image: str | None = None,
     from_path: Path | None = None,
+    net: bool = False,
     allow_cidrs: Sequence[str] = (),
     env: Mapping[str, str] | None = None,
+    ports: Sequence[tuple[int, int]] = (),
+    volumes: Sequence[tuple[str, str, bool]] = (),
 ) -> None:
     """`smolvm machine create --name NAME [--image IMG | --from PATH]
     [--allow-cidr CIDR ...] [-e K=V ...]`. NAME is passed as
@@ -131,19 +134,24 @@ def machine_create(
     no-pull-at-start property. The flag form gives the same
     result without the Smolfile complication.
 
-    `--net` is sent explicitly when `allow_cidrs` is non-empty.
-    `--allow-cidr` implies `--net` per the CLI help, but sending
-    `--net` explicitly is harmless and ensures the guest has
+    `--net` is sent explicitly when `net=True` or `allow_cidrs` is
+    non-empty. `--allow-cidr` implies `--net` per the CLI help, but
+    sending `--net` explicitly is harmless and ensures the guest has
     network access even if that implication changes across versions."""
     args: list[str] = ["machine", "create", "--name", name]
     if image is not None:
         args += ["--image", image]
     if from_path is not None:
         args += ["--from", str(from_path)]
-    if allow_cidrs:
+    if net or allow_cidrs:
         args.append("--net")
     for cidr in allow_cidrs:
         args += ["--allow-cidr", cidr]
+    for host_port, guest_port in ports:
+        args += ["-p", f"{host_port}:{guest_port}"]
+    for host_path, guest_path, read_only in volumes:
+        suffix = ":ro" if read_only else ""
+        args += ["-v", f"{host_path}:{guest_path}{suffix}"]
     if env:
         for k, v in env.items():
             args += ["-e", f"{k}={v}"]

bot_bottle/cli/start.py

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

bot_bottle/config_store.py

@@ -0,0 +1,71 @@
+"""SQLite-backed bot-bottle configuration store."""
+
+from __future__ import annotations
+
+from pathlib import Path
+
+try:
+    from .db_store import DbStore
+    from .migrations import TableMigrations
+    from .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

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

bot_bottle/store_manager.py

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

docs/prds/prd-new-smolmachines-sidecar-vm.md

@@ -0,0 +1,115 @@
+# PRD prd-new: smolmachines sidecar VM
+
+- **Status:** Active
+- **Author:** codex
+- **Created:** 2026-07-09
+- **Issue:** #332
+
+## Summary
+
+Run the smolmachines backend's trusted sidecar bundle as its own smolVM instead
+of a Docker container. A bottle then consists of an agent VM plus a sidecar VM,
+with the agent VM's TSI allowlist limited to the per-bottle agent-facing
+sidecar surface.
+
+## Problem
+
+The smolmachines backend currently runs the agent in smolVM but keeps
+egress/git-gate/supervise in a Docker sidecar bundle. That hybrid launch path
+keeps Docker in the trusted runtime path and leaves smolmachines with a
+different isolation boundary than its agent VM design suggests.
+
+The existing Docker sidecar path also uses Docker port publishing to bind only
+agent-facing services to the per-bottle host address. TSI is IP-only, so the
+smolVM replacement must preserve that scoping: publishing sidecar services on
+generic host localhost would let the agent reach unrelated host services or
+other bottle sidecars if those services share the allowed address.
+
+## Goals / Success Criteria
+
+1. `BOT_BOTTLE_BACKEND=smolmachines ./cli.py start <agent>` runs with both the
+   agent and sidecar as smolVMs.
+2. The smolmachines launch path no longer uses `docker run` for the sidecar
+   bundle.
+3. Agent-facing egress, git-gate, and supervise endpoints are exposed only
+   through the bottle's own published sidecar surface.
+4. Internal sidecar-only ports are not reachable from the agent VM.
+5. The agent VM TSI allowlist remains fail-closed and limited to the per-bottle
+   address.
+6. Sidecar config, secrets, and state are delivered to the sidecar VM without
+   exposing provider or forge credentials to the agent VM.
+7. Teardown removes both VMs and any host-side published port state.
+
+## Non-goals
+
+- Rewriting the egress, git-gate, or supervise daemons.
+- Changing the Docker backend's sidecar bundle behavior.
+- Replacing the existing agent image build and pack pipeline except where shared
+  helper extraction is needed for sidecar image packing.
+- Weakening the current smolmachines TSI allowlist checks.
+
+## Design
+
+The sidecar VM continues to use the existing sidecar bundle image and
+`/app/sidecar_init.py` supervisor. The launch flow builds the sidecar image,
+packs it into a `.smolmachine` artifact, creates a per-bottle sidecar VM from
+that artifact, passes the same daemon-selection environment the Docker bundle
+uses today, mounts or copies the same daemon-private config/state inputs, starts
+the sidecar VM, wraps the sidecar VM's raw smolVM-published ports with a local
+per-bottle forwarder, then starts the agent VM with
+`--allow-cidr <per-bottle-address>/32`.
+
+smolVM currently exposes guest ports as `HOST:GUEST` port pairs bound on host
+loopback. It does not expose an address-scoped bind like
+`127.0.0.16:<port>:9099`. Because TSI is IP-only, bot-bottle must not advertise
+those raw loopback ports directly to the agent. Instead, bot-bottle runs one
+small stdlib Python TCP forwarder process per bottle:
+
+```text
+agent VM
+  -> TSI allows only <bottle-address>/32
+  -> bot-bottle forwarder bound to <bottle-address>:<random-port>
+  -> raw smolVM-published sidecar port on 127.0.0.1:<raw-port>
+  -> sidecar VM service
+```
+
+Agent-facing URLs are stamped from the forwarder ports, never the raw smolVM
+ports:
+
+- `HTTP_PROXY` / `HTTPS_PROXY` point at the published egress proxy port.
+- `GIT_GATE_URL` points at the published git HTTP port when git-gate is enabled.
+- `MCP_SUPERVISE_URL` points at the published supervise port when supervise is
+  enabled.
+
+Internal sidecar-only services stay bound inside the sidecar VM and are not
+published.
+
+### Forwarder constraints
+
+- Reject wildcard listener binds (`0.0.0.0`, `::`) and generic localhost
+  listener binds (`127.0.0.1`, `::1`) for agent-facing forwards.
+- Accept only fixed loopback targets selected by launch; the client cannot
+  choose or influence the forwarding target.
+- Forward bytes transparently without parsing HTTP, CONNECT, TLS, Git, or MCP.
+- Log lifecycle and endpoint metadata only; never log payload bytes.
+- Create forwards only for egress, git HTTP when enabled, and supervise when
+  enabled.
+- Fail closed if any listener cannot bind exactly the requested per-bottle
+  address.
+
+## Implementation chunks
+
+1. Extract the existing image-to-smolmachine pack helper so agent and sidecar
+   artifacts share the same cache and registry path.
+2. Add a strict stdlib host TCP forwarder for per-bottle address-bound
+   forwarding to raw smolVM-published sidecar ports.
+3. Add a sidecar VM launch spec and lifecycle helpers: pack, create, start,
+   publish/discover ports, stop, delete.
+4. Switch smolmachines launch from Docker sidecar bundle lifecycle to sidecar VM
+   plus forwarder lifecycle.
+5. Update egress apply / reload paths for the sidecar VM supervisor.
+6. Add unit tests for argv shape, URL stamping, teardown, and fail-closed
+   behavior.
+7. Add or update integration coverage for agent-to-sidecar reachability, host
+   localhost denial, other-bottle alias denial, internal port denial, and
+   teardown cleanup.

tests/unit/test_cli_start_headless.py

@@ -184,6 +184,18 @@ 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,6 +57,12 @@ 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)
@@ -66,6 +72,7 @@ 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()
 
     # ------------------------------------------------------------------
@@ -124,6 +131,19 @@ class TestCmdStartSelector(unittest.TestCase):
         spec = self._launch_mock.call_args[0][0]
         self.assertEqual(("claude", "dev"), spec.bottle_names)
 
+    def test_image_policy_forwarded_to_spec(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value="cached"):
+            start_mod.cmd_start(["researcher"])
+        self._launch_mock.assert_called_once()
+        spec = self._launch_mock.call_args[0][0]
+        self.assertEqual("cached", spec.image_policy)
+
+    def test_image_policy_cancel_returns_0(self):
+        with patch("bot_bottle.cli.start._select_image_policy", return_value=None):
+            rc = start_mod.cmd_start(["researcher"])
+        self.assertEqual(0, rc)
+        self._launch_mock.assert_not_called()
+
     def test_empty_bottle_selection_forwarded(self):
         self._bottle_picker_mock.return_value = []
         start_mod.cmd_start(["researcher"])
@@ -215,6 +235,7 @@ class TestCmdStartLabelCollision(unittest.TestCase):
         ).start()
         # Stub the bottle picker to always return a selection.
         patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start()
+        patch("bot_bottle.cli.start._select_image_policy", return_value="fresh").start()
         self.addCleanup(patch.stopall)
 
     def test_no_collision_proceeds_without_reprompt(self):

tests/unit/test_cli_start_stale.py

@@ -0,0 +1,164 @@
+"""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,6 +118,7 @@ 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."""
@@ -148,6 +149,7 @@ def _plan(
         agent_name="demo",
         copy_cwd=False,
         user_cwd="/tmp/x",
+        image_policy=image_policy,
     )
     return DockerBottlePlan(
         spec=spec,
@@ -265,7 +267,8 @@ 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.
-        for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
+        cases: list[dict[str, Any]] = [{}, {"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"])
@@ -300,6 +303,11 @@ 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

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

tests/unit/test_docker_launch_committed_image.py

@@ -3,6 +3,7 @@
 from __future__ import annotations
 
 import contextlib
+import dataclasses
 import io
 import tempfile
 import unittest
@@ -16,6 +17,7 @@ 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
 
 
@@ -180,6 +182,24 @@ 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,6 +9,7 @@ from __future__ import annotations
 
 import subprocess
 import unittest
+from datetime import timezone
 from unittest.mock import patch
 
 from bot_bottle.backend.docker import util as docker_mod
@@ -54,6 +55,51 @@ 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

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

tests/unit/test_macos_container_launch.py

@@ -2,6 +2,7 @@
 
 from __future__ import annotations
 
+import dataclasses
 import unittest
 import tempfile
 from pathlib import Path
@@ -15,6 +16,7 @@ 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({
@@ -81,7 +83,7 @@ def _plan(
         provisioned_env={"CODEX_HOME": "/run/codex-home"},
     )
     return cast(MacosContainerBottlePlan, SimpleNamespace(
-        spec=SimpleNamespace(),
+        spec=SimpleNamespace(image_policy="fresh"),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         slug="dev-abc",
@@ -292,7 +294,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
 
 def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
     return MacosContainerBottlePlan(
-        spec=cast(BottleSpec, SimpleNamespace()),
+        spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
         manifest=_MANIFEST,
         stage_dir=stage_dir,
         git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -365,5 +367,82 @@ 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,5 +273,71 @@ 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_egress_apply.py

@@ -0,0 +1,58 @@
+from __future__ import annotations
+
+import unittest
+from unittest.mock import patch
+
+from bot_bottle.backend.egress_apply import EgressApplyError
+from bot_bottle.backend.smolmachines import egress_apply
+from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
+
+
+class TestFetchCurrentRoutes(unittest.TestCase):
+    def test_reads_routes_from_sidecar_vm(self):
+        with patch.object(
+            egress_apply._smolvm,
+            "machine_exec",
+            return_value=SmolvmRunResult(0, "routes", ""),
+        ) as exec_:
+            self.assertEqual("routes", egress_apply.fetch_current_routes("dev-abc"))
+        exec_.assert_called_once_with(
+            "bot-bottle-sidecars-dev-abc",
+            ["cat", "/etc/egress/routes.yaml"],
+        )
+
+    def test_read_failure_raises_apply_error(self):
+        with patch.object(
+            egress_apply._smolvm,
+            "machine_exec",
+            return_value=SmolvmRunResult(1, "", "nope"),
+        ):
+            with self.assertRaises(EgressApplyError):
+                egress_apply.fetch_current_routes("dev-abc")
+
+
+class TestReload(unittest.TestCase):
+    def test_sends_hup_to_sidecar_init(self):
+        with patch.object(
+            egress_apply._smolvm,
+            "machine_exec",
+            return_value=SmolvmRunResult(0, "", ""),
+        ) as exec_:
+            egress_apply.SmolmachinesEgressApplicator()._signal_bundle_reload("dev-abc")
+        exec_.assert_called_once_with(
+            "bot-bottle-sidecars-dev-abc",
+            ["sh", "-c", "kill -HUP 1"],
+        )
+
+    def test_reload_failure_raises_apply_error(self):
+        with patch.object(
+            egress_apply._smolvm,
+            "machine_exec",
+            return_value=SmolvmRunResult(1, "", "nope"),
+        ):
+            with self.assertRaises(EgressApplyError):
+                egress_apply.SmolmachinesEgressApplicator()._signal_bundle_reload("dev-abc")
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_enumerate.py

@@ -0,0 +1,66 @@
+"""Unit: smolmachines active-agent enumeration."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+import unittest
+from unittest.mock import patch
+
+from bot_bottle.backend.smolmachines import enumerate as _enumerate
+
+
+def _completed(
+    stdout: str,
+    returncode: int = 0,
+) -> subprocess.CompletedProcess:  # type: ignore[type-arg]
+    return subprocess.CompletedProcess(
+        args=[],
+        returncode=returncode,
+        stdout=stdout,
+        stderr="",
+    )
+
+
+class TestEnumerateActive(unittest.TestCase):
+    def test_skips_sidecar_vms(self):
+        machines = [
+            {"name": "bot-bottle-demo-abc12", "state": "running"},
+            {"name": "bot-bottle-sidecars-demo-abc12", "state": "running"},
+            {"name": "bot-bottle-stopped-xyz", "state": "stopped"},
+            {"name": "other", "state": "running"},
+        ]
+
+        class _Metadata:
+            agent_name = "demo"
+            started_at = "2026-07-09T00:00:00Z"
+            label = "Demo"
+            color = "green"
+
+        with patch(
+            "bot_bottle.backend.smolmachines.enumerate.subprocess.run",
+            return_value=_completed(json.dumps(machines)),
+        ), patch(
+            "bot_bottle.backend.smolmachines.enumerate._query_bundle_services",
+            return_value={"demo-abc12": ("egress",)},
+        ), patch(
+            "bot_bottle.backend.smolmachines.enumerate.read_metadata",
+            return_value=_Metadata(),
+        ) as read_metadata:
+            active = _enumerate.enumerate_active()
+
+        self.assertEqual(1, len(active))
+        self.assertEqual("demo-abc12", active[0].slug)
+        self.assertEqual(("egress",), active[0].services)
+        read_metadata.assert_called_once_with("demo-abc12")
+
+    def test_invalid_smolvm_output_returns_empty(self):
+        with patch(
+            "bot_bottle.backend.smolmachines.enumerate.subprocess.run",
+            return_value=_completed("not json"),
+        ):
+            self.assertEqual([], _enumerate.enumerate_active())
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_launch_image.py

@@ -187,6 +187,54 @@ 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_port_forward.py

@@ -0,0 +1,325 @@
+from __future__ import annotations
+
+import json
+import socket
+import subprocess
+import threading
+import unittest
+from unittest.mock import MagicMock, patch
+
+from bot_bottle.backend.smolmachines.port_forward import (
+    ForwardSpec,
+    ForwarderHandle,
+    _CompletedProcess,
+    _Forwarder,
+    _pipe,
+    main,
+    start_forwarder,
+    stop_forwarder,
+)
+
+
+def _echo_server() -> tuple[socket.socket, int]:
+    listener = socket.create_server(("127.0.0.1", 0))
+    port = int(listener.getsockname()[1])
+
+    def run() -> None:
+        with listener:
+            conn, _ = listener.accept()
+            with conn:
+                while True:
+                    data = conn.recv(65536)
+                    if not data:
+                        return
+                    conn.sendall(data)
+
+    threading.Thread(target=run, daemon=True).start()
+    return listener, port
+
+
+class TestForwarderValidation(unittest.TestCase):
+    def test_empty_specs_returns_noop_handle(self):
+        handle = start_forwarder(())
+
+        self.assertEqual((), handle.forwards)
+        self.assertEqual(0, handle.process.poll())
+        stop_forwarder(handle)
+
+    def test_rejects_generic_localhost_listener(self):
+        with self.assertRaises(SystemExit):
+            start_forwarder((
+                ForwardSpec(
+                    label="egress",
+                    listen_host="127.0.0.1",
+                    listen_port=0,
+                    target_host="127.0.0.1",
+                    target_port=9099,
+                ),
+            ))
+
+    def test_rejects_wildcard_listener(self):
+        with self.assertRaises(SystemExit):
+            start_forwarder((
+                ForwardSpec(
+                    label="egress",
+                    listen_host="0.0.0.0",
+                    listen_port=0,
+                    target_host="127.0.0.1",
+                    target_port=9099,
+                ),
+            ))
+
+    def test_rejects_non_loopback_target(self):
+        with self.assertRaises(SystemExit):
+            start_forwarder((
+                ForwardSpec(
+                    label="egress",
+                    listen_host="127.0.0.16",
+                    listen_port=0,
+                    target_host="192.0.2.10",
+                    target_port=9099,
+                ),
+            ))
+
+    def test_rejects_invalid_ports(self):
+        with self.assertRaises(SystemExit):
+            start_forwarder((
+                ForwardSpec("bad-listen", "127.0.0.2", 70000, "127.0.0.1", 9099),
+            ))
+        with self.assertRaises(SystemExit):
+            start_forwarder((
+                ForwardSpec("bad-target", "127.0.0.2", 0, "127.0.0.1", 0),
+            ))
+
+    def test_completed_process_methods_are_noops(self):
+        proc = _CompletedProcess()
+
+        proc.terminate()
+        proc.kill()
+
+        self.assertEqual(0, proc.poll())
+        self.assertEqual(0, proc.wait(timeout=1))
+
+
+class _FakeStdout:
+    def __init__(self, line: str):
+        self.line = line
+        self.closed = False
+
+    def readline(self) -> str:
+        return self.line
+
+    def close(self) -> None:
+        self.closed = True
+
+
+class TestForwarderProcess(unittest.TestCase):
+    def test_start_parses_ready_payload(self):
+        payload = {
+            "forwards": [{
+                "label": "egress",
+                "listen_host": "127.0.0.2",
+                "listen_port": 61000,
+                "target_host": "127.0.0.1",
+                "target_port": 55000,
+            }],
+        }
+        proc = MagicMock()
+        proc.stdout = _FakeStdout(json.dumps(payload) + "\n")
+
+        with patch(
+            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
+            return_value=proc,
+        ) as popen:
+            handle = start_forwarder((
+                ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
+            ))
+
+        self.assertIs(proc, handle.process)
+        self.assertEqual(61000, handle.forwards[0].listen_port)
+        self.assertTrue(proc.stdout.closed)
+        argv = popen.call_args.args[0]
+        self.assertIn("--spec-json", argv)
+
+    def test_start_dies_when_helper_exits_before_ready(self):
+        proc = MagicMock()
+        proc.stdout = _FakeStdout("")
+
+        with patch(
+            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
+            return_value=proc,
+        ):
+            with self.assertRaises(SystemExit):
+                start_forwarder((
+                    ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
+                ))
+
+        proc.terminate.assert_called_once()
+        proc.wait.assert_called_once_with(timeout=5)
+
+    def test_start_dies_on_invalid_ready_payload(self):
+        proc = MagicMock()
+        proc.stdout = _FakeStdout('{"wrong": []}\n')
+
+        with patch(
+            "bot_bottle.backend.smolmachines.port_forward.subprocess.Popen",
+            return_value=proc,
+        ):
+            with self.assertRaises(SystemExit):
+                start_forwarder((
+                    ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 55000),
+                ))
+
+        proc.terminate.assert_called_once()
+        proc.wait.assert_called_once_with(timeout=5)
+
+    def test_stop_kills_process_after_timeout(self):
+        proc = MagicMock()
+        proc.poll.return_value = None
+        proc.wait.side_effect = [subprocess.TimeoutExpired("cmd", 5), 0]
+
+        stop_forwarder(ForwarderHandle(process=proc, forwards=()))
+
+        proc.terminate.assert_called_once()
+        proc.kill.assert_called_once()
+        self.assertEqual(2, proc.wait.call_count)
+
+
+class TestForwarding(unittest.TestCase):
+    def test_forwards_bytes_both_directions(self):
+        try:
+            probe = socket.create_server(("127.0.0.2", 0))
+        except OSError:
+            self.skipTest("127.0.0.2 is not bindable on this host")
+        else:
+            probe.close()
+        _listener, target_port = _echo_server()
+        handle = start_forwarder((
+            ForwardSpec(
+                label="egress",
+                listen_host="127.0.0.2",
+                listen_port=0,
+                target_host="127.0.0.1",
+                target_port=target_port,
+            ),
+        ))
+        try:
+            bound = handle.forwards[0]
+            with socket.create_connection((bound.listen_host, bound.listen_port)) as sock:
+                sock.sendall(b"hello")
+                self.assertEqual(b"hello", sock.recv(5))
+        finally:
+            stop_forwarder(handle)
+
+
+class TestForwarderInProcess(unittest.TestCase):
+    def test_forwarder_serves_and_stops_in_process(self):
+        _listener, target_port = _echo_server()
+        forwarder = _Forwarder((
+            ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", target_port),
+        ))
+        forwarder.start()
+        thread = threading.Thread(target=forwarder.serve, daemon=True)
+        thread.start()
+        try:
+            bound = forwarder.bound[0]
+            with socket.create_connection((bound.listen_host, bound.listen_port)) as sock:
+                sock.sendall(b"ping")
+                self.assertEqual(b"ping", sock.recv(4))
+        finally:
+            forwarder.stop()
+            thread.join(timeout=2)
+
+        self.assertFalse(thread.is_alive())
+
+    def test_forwarder_logs_target_connect_failure(self):
+        forwarder = _Forwarder((
+            ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 1),
+        ))
+        forwarder.start()
+        thread = threading.Thread(target=forwarder.serve, daemon=True)
+        thread.start()
+        try:
+            bound = forwarder.bound[0]
+            with patch("bot_bottle.backend.smolmachines.port_forward.warn") as warn:
+                with socket.create_connection((bound.listen_host, bound.listen_port)):
+                    pass
+                for _ in range(100):
+                    if warn.called:
+                        break
+                    threading.Event().wait(0.01)
+            self.assertTrue(warn.called)
+        finally:
+            forwarder.stop()
+            thread.join(timeout=2)
+
+    def test_pipe_stops_when_recv_fails(self):
+        src, dst = socket.socketpair()
+        src.close()
+        try:
+            _pipe(src, dst)
+        finally:
+            dst.close()
+
+    def test_pipe_stops_when_send_fails(self):
+        src, src_peer = socket.socketpair()
+        _dst_peer, dst = socket.socketpair()
+        dst.close()
+        try:
+            src_peer.sendall(b"hello")
+            src_peer.shutdown(socket.SHUT_WR)
+            _pipe(src, dst)
+        finally:
+            src.close()
+            src_peer.close()
+            _dst_peer.close()
+
+
+class TestMain(unittest.TestCase):
+    def test_main_starts_serves_and_stops_forwarder(self):
+        calls: list[str] = []
+        handlers = []
+
+        class FakeForwarder:
+            def __init__(self, specs):  # type: ignore[no-untyped-def]
+                self.specs = specs
+                self.bound = specs
+
+            def start(self) -> None:
+                calls.append("start")
+
+            def serve(self) -> None:
+                calls.append("serve")
+
+            def stop(self) -> None:
+                calls.append("stop")
+
+        def record_signal(_signum, handler):  # type: ignore[no-untyped-def]
+            handlers.append(handler)
+
+        spec = ForwardSpec("egress", "127.0.0.2", 0, "127.0.0.1", 9099)
+        with patch(
+            "bot_bottle.backend.smolmachines.port_forward._Forwarder",
+            FakeForwarder,
+        ), patch(
+            "bot_bottle.backend.smolmachines.port_forward.signal.signal",
+            side_effect=record_signal,
+        ), patch("builtins.print") as print_mock:
+            self.assertEqual(
+                0,
+                main(("--spec-json", json.dumps([{
+                    "label": spec.label,
+                    "listen_host": spec.listen_host,
+                    "listen_port": spec.listen_port,
+                    "target_host": spec.target_host,
+                    "target_port": spec.target_port,
+                }]))),
+            )
+            handlers[0](15, None)
+
+        self.assertEqual(["start", "serve", "stop", "stop"], calls)
+        self.assertIn("forwards", print_mock.call_args.args[0])
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_provision.py

@@ -9,6 +9,7 @@ from __future__ import annotations
 import subprocess
 import tempfile
 import unittest
+from contextlib import ExitStack
 from dataclasses import replace
 from pathlib import Path
 from unittest.mock import MagicMock, patch
@@ -27,6 +28,7 @@ from bot_bottle.backend.smolmachines.bottle_plan import (
     SmolmachinesBottlePlan,
 )
 from bot_bottle.backend.smolmachines import launch as _launch
+from bot_bottle.backend.smolmachines import port_forward as _forward
 from bot_bottle.backend.smolmachines.launch import _bundle_launch_spec
 from bot_bottle.backend.util import AGENT_CA_PATH
 from bot_bottle.egress import EgressPlan, EgressRoute
@@ -432,12 +434,7 @@ class TestBundleLaunchSpec(unittest.TestCase):
 
     def test_canary_env_visible_to_smolvm_guest(self):
         plan = _plan(canary=True)
-        with patch.object(
-            _launch._bundle,
-            "bundle_host_port",
-            return_value="65000",
-        ):
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
+        stamped = _launch._discover_urls(plan, "127.0.0.16", {9099: 65000})
 
         self.assertEqual(
             "fake-canary-value",
@@ -506,23 +503,111 @@ class TestProvisionGitUser(unittest.TestCase):
         self.assertIn("bot@example.com", calls[0][0])
 
 
-class TestProxyHost(unittest.TestCase):
-    """_proxy_host returns the bridge gateway on Linux and the loopback
-    alias on other platforms."""
-
-    def test_linux_returns_bundle_gateway(self):
+class TestLaunchResourceWiring(unittest.TestCase):
+    def test_allocate_resources_uses_loopback_alias_and_bundle_name(self):
         plan = _plan()
-        with patch("bot_bottle.backend.smolmachines.launch.platform.system",
-                   return_value="Linux"):
-            result = _launch._proxy_host(plan, "127.0.0.16")
-        self.assertEqual(plan.bundle_gateway, result)
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._loopback.ensure_pool",
+        ) as ensure_pool, patch(
+            "bot_bottle.backend.smolmachines.launch._loopback.allocate",
+            return_value="127.0.0.16",
+        ) as allocate:
+            loopback_ip, network = _launch._allocate_resources(plan, ExitStack())
 
-    def test_non_linux_returns_loopback(self):
+        ensure_pool.assert_called_once_with()
+        allocate.assert_called_once_with("demo-abc12")
+        self.assertEqual("127.0.0.16", loopback_ip)
+        self.assertEqual("bot-bottle-bundle-demo-abc12", network)
+
+    def test_start_bundle_wraps_raw_vm_ports_with_forwarder_ports(self):
+        plan = _plan(supervise=True)
+        plan = replace(
+            plan,
+            git_gate_plan=replace(
+                plan.git_gate_plan,
+                upstreams=(GitGateUpstream(
+                    name="bot-bottle",
+                    upstream_url="ssh://git@host/repo.git",
+                    upstream_host="host",
+                    upstream_port="22",
+                    identity_file="/tmp/key",
+                    known_host_key="",
+                ),),
+            ),
+        )
+        stack = MagicMock(spec=ExitStack)
+        raw_launch = MagicMock(raw_ports={9099: 55001, 9420: 55002, 9100: 55003})
+        handle = _forward.ForwarderHandle(
+            process=MagicMock(),
+            forwards=(
+                _forward.ForwardSpec("egress", "127.0.0.16", 65001, "127.0.0.1", 55001),
+                _forward.ForwardSpec("git-http", "127.0.0.16", 65002, "127.0.0.1", 55002),
+                _forward.ForwardSpec("supervise", "127.0.0.16", 65003, "127.0.0.1", 55003),
+            ),
+        )
+
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._provision_git_gate_keys",
+            return_value=plan,
+        ), patch(
+            "bot_bottle.backend.smolmachines.launch._ensure_smolmachine",
+            return_value=Path("/cache/sidecar.smolmachine"),
+        ) as ensure, patch(
+            "bot_bottle.backend.smolmachines.launch._bundle.start_bundle_vm",
+            return_value=raw_launch,
+        ) as start_vm, patch(
+            "bot_bottle.backend.smolmachines.launch._forward.start_forwarder",
+            return_value=handle,
+        ) as start_forwarder:
+            stamped = _launch._start_bundle(plan, "net", "127.0.0.16", stack)
+
+        ensure.assert_called_once()
+        start_vm.assert_called_once()
+        self.assertEqual(Path("/cache/sidecar.smolmachine"), start_vm.call_args.kwargs["from_path"])
+        specs = start_forwarder.call_args.args[0]
+        self.assertEqual(
+            (
+                _forward.ForwardSpec("egress", "127.0.0.16", 0, "127.0.0.1", 55001),
+                _forward.ForwardSpec("git-http", "127.0.0.16", 0, "127.0.0.1", 55002),
+                _forward.ForwardSpec("supervise", "127.0.0.16", 0, "127.0.0.1", 55003),
+            ),
+            specs,
+        )
+        self.assertEqual("http://127.0.0.16:65001", stamped.agent_proxy_url)
+        self.assertEqual("127.0.0.16:65002", stamped.agent_git_gate_host)
+        self.assertEqual("http://127.0.0.16:65003/", stamped.agent_supervise_url)
+        self.assertEqual(2, stack.callback.call_count)
+
+    def test_init_vm_repairs_ownership_then_waits_for_exec(self):
         plan = _plan()
-        with patch("bot_bottle.backend.smolmachines.launch.platform.system",
-                   return_value="Darwin"):
-            result = _launch._proxy_host(plan, "127.0.0.16")
-        self.assertEqual("127.0.0.16", result)
+        with patch(
+            "bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
+        ) as machine_exec, patch(
+            "bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
+        ) as wait_exec_ready:
+            _launch._init_vm(plan)
+
+        machine_exec.assert_called_once()
+        argv = machine_exec.call_args.args[1]
+        self.assertEqual(["sh", "-c"], argv[:2])
+        self.assertIn("chown -R node:node /home/node", argv[2])
+        wait_exec_ready.assert_called_once_with(plan.machine_name)
+
+
+class TestPortLabels(unittest.TestCase):
+    def test_known_and_dynamic_port_labels_round_trip(self):
+        for port, label in (
+            (9099, "egress"),
+            (9420, "git-http"),
+            (9100, "supervise"),
+            (12345, "port-12345"),
+        ):
+            self.assertEqual(label, _launch._label_for_port(port))
+            self.assertEqual(port, _launch._port_for_label(label))
+
+    def test_unknown_label_is_rejected(self):
+        with self.assertRaises(ValueError):
+            _launch._port_for_label("sidecar")
 
 
 class TestDiscoverUrls(unittest.TestCase):
@@ -544,14 +629,20 @@ class TestDiscoverUrls(unittest.TestCase):
                 ),),
             ),
         )
-        with patch.object(_launch._bundle, "bundle_host_port", return_value="9420"):
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
-        self.assertEqual("127.0.0.16:9420", stamped.agent_git_gate_host)
+        stamped = _launch._discover_urls(
+            plan,
+            "127.0.0.16",
+            {9099: 65000, 9420: 65001},
+        )
+        self.assertEqual("127.0.0.16:65001", stamped.agent_git_gate_host)
 
     def test_supervise_url_set_when_supervise_present(self):
         plan = _plan(supervise=True)
-        with patch.object(_launch._bundle, "bundle_host_port", return_value="55556"):
-            stamped = _launch._discover_urls(plan, "127.0.0.16")
+        stamped = _launch._discover_urls(
+            plan,
+            "127.0.0.16",
+            {9099: 65000, 9100: 55556},
+        )
         self.assertEqual("http://127.0.0.16:55556/", stamped.agent_supervise_url)
 
 

tests/unit/test_smolmachines_sidecar_bundle.py

@@ -14,13 +14,16 @@ from unittest.mock import patch
 
 from bot_bottle.backend.smolmachines.sidecar_bundle import (
     BundleLaunchSpec,
+    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,
 )
 
 
@@ -200,6 +203,66 @@ class TestEnsureBundleImage(unittest.TestCase):
         self.assertEqual("Dockerfile.sidecars", kwargs["dockerfile"])
 
 
+class TestStartBundleVm(unittest.TestCase):
+    def test_allocate_raw_host_ports_returns_bindable_ports(self):
+        raw_ports = allocate_raw_host_ports((9099, 9420))
+
+        self.assertEqual({9099, 9420}, set(raw_ports))
+        for host_port in raw_ports.values():
+            self.assertIsInstance(host_port, int)
+            self.assertGreater(host_port, 0)
+        self.assertNotEqual(raw_ports[9099], raw_ports[9420])
+
+    def test_creates_sidecar_vm_with_ports_volumes_and_env(self):
+        calls: list[tuple[str, tuple[object, ...], dict[str, object]]] = []
+
+        def record(name):  # type: ignore
+            def inner(*args, **kwargs):  # type: ignore
+                calls.append((name, args, kwargs))
+            return inner
+
+        with patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle.allocate_raw_host_ports",
+            return_value={9099: 55001, 9420: 55002},
+        ), patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_create",
+            side_effect=record("create"),
+        ), patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_start",
+            side_effect=record("start"),
+        ), patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.wait_exec_ready",
+            side_effect=record("wait"),
+        ):
+            launch = start_bundle_vm(
+                _spec(
+                    daemons_csv="egress,git-http",
+                    environment=("FOO=bar", "TOKEN"),
+                    volumes=(("/host/routes", "/etc/egress", True),),
+                    ports_to_publish=(9099, 9420),
+                ),
+                from_path=Path("/cache/sidecar.smolmachine"),
+                host_env={"TOKEN": "secret"},
+            )
+
+        self.assertEqual({9099: 55001, 9420: 55002}, launch.raw_ports)
+        create = calls[0]
+        self.assertEqual("create", create[0])
+        self.assertEqual(("bot-bottle-sidecars-demo-abc12",), create[1])
+        self.assertEqual(Path("/cache/sidecar.smolmachine"), create[2]["from_path"])
+        self.assertTrue(create[2]["net"])
+        self.assertEqual(((55001, 9099), (55002, 9420)), create[2]["ports"])
+        self.assertEqual((("/host/routes", "/etc/egress", True),), create[2]["volumes"])
+        self.assertEqual(
+            {
+                "BOT_BOTTLE_SIDECAR_DAEMONS": "egress,git-http",
+                "FOO": "bar",
+                "TOKEN": "secret",
+            },
+            create[2]["env"],
+        )
+
+
 class TestStopBundle(unittest.TestCase):
     def _patch_run(self, **kwargs):  # type: ignore
         return patch(
@@ -222,5 +285,38 @@ class TestStopBundle(unittest.TestCase):
             stop_bundle("demo-abc12")  # no raise
 
 
+class TestStopBundleVm(unittest.TestCase):
+    def test_stops_then_deletes_sidecar_vm(self):
+        with patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
+        ) as stop, patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
+        ) as delete:
+            stop_bundle_vm("demo-abc12")
+        stop.assert_called_once_with("bot-bottle-sidecars-demo-abc12")
+        delete.assert_called_once_with("bot-bottle-sidecars-demo-abc12")
+
+    def test_stop_and_delete_errors_are_warnings(self):
+        from bot_bottle.backend.smolmachines.smolvm import SmolvmError
+        result = subprocess.CompletedProcess(
+            args=[],
+            returncode=1,
+            stdout="",
+            stderr="failed",
+        )
+        with patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_stop",
+            side_effect=SmolvmError(["smolvm", "machine", "stop"], result),
+        ), patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle._smolvm.machine_delete",
+            side_effect=SmolvmError(["smolvm", "machine", "delete"], result),
+        ), patch(
+            "bot_bottle.backend.smolmachines.sidecar_bundle.warn",
+        ) as warn:
+            stop_bundle_vm("demo-abc12")
+
+        self.assertEqual(2, warn.call_count)
+
+
 if __name__ == "__main__":
     unittest.main()

tests/unit/test_smolmachines_smolvm.py

@@ -106,6 +106,22 @@ class TestArgvShapes(unittest.TestCase):
         self.assertIn("--name", argv)
         self.assertIn("agent-xyz", argv)
 
+    def test_machine_create_with_net_ports_and_volumes(self):
+        with self._patch_run() as m:
+            machine_create(
+                "sidecar-xyz",
+                from_path=Path("/stage/sidecar.smolmachine"),
+                net=True,
+                ports=((55000, 9099),),
+                volumes=(("/host/routes", "/etc/egress", True),),
+            )
+        argv = m.call_args.args[0]
+        self.assertIn("--net", argv)
+        self.assertIn("-p", argv)
+        self.assertIn("55000:9099", argv)
+        self.assertIn("-v", argv)
+        self.assertIn("/host/routes:/etc/egress:ro", argv)
+
     def test_machine_create_omits_net_when_no_allow_cidrs(self):
         with self._patch_run() as m:
             machine_create("agent-xyz", from_path=Path("/x.smolmachine"))

tests/unit/test_stale_checks.py

@@ -0,0 +1,302 @@
+"""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()