bot-bottle/bot_bottle/backend/smolmachines/launch.py

"""End-to-end launch flow for the smolmachines backend.

Builds the sidecar bundle smolmachine, starts it as a sidecar VM
with real daemons + their config files, creates + starts the agent
smolVM, yields a `SmolmachinesBottle` handle, and tears everything
down on context exit.

The bundle's daemons consume the inner Plans the docker backend
already produces: egress reads routes + CAs from the EgressPlan.
Git-gate + supervise plumb through the same plans the docker
backend uses, minus the docker-network fields that don't apply here."""

from __future__ import annotations

import dataclasses
import os
import shutil
from contextlib import ExitStack, contextmanager
from pathlib import Path
from typing import Callable, Generator

from ...egress import (
    egress_agent_env_entries,
    egress_resolve_token_values,
    egress_sidecar_env_entries,
)
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde
from ..docker import util as docker_mod
from ..docker.egress import (
    EGRESS_PORT as _EGRESS_PORT,
    egress_tls_init,
)
from ...git_gate import (
    provision_git_gate_dynamic_keys,
    revoke_git_gate_provisioned_keys,
)
from ...log import 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
from .bottle_plan import SmolmachinesBottlePlan
from .local_registry import crane_push_tarball, ephemeral_registry


# Repo root, used as the `docker build` context for the agent image.
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)


# Single virtiofs mount for egress + git-gate files. libkrun limits
# the total of mounts + port-mappings to 5; with 3 daemon ports the
# sidecar VM can carry at most 2 mounts. Egress CA/routes and
# git-gate scripts/creds are staged into subdirectories of one host
# dir and mounted here. Env vars (EGRESS_CONFDIR, EGRESS_ROUTES,
# and the Dockerfile's git-gate wrapper) point each daemon at its
# subdirectory.
_SIDECAR_DATA_DIR_IN_VM = "/bot-bottle-data"
_EGRESS_CONFDIR_IN_VM = f"{_SIDECAR_DATA_DIR_IN_VM}/egress"
_GIT_GATE_SCRIPTS_DIR_IN_VM = f"{_SIDECAR_DATA_DIR_IN_VM}/git-gate"


# Per-host cache for `smolvm pack create` outputs. Keyed by the
# docker image ID so a Dockerfile change automatically invalidates
# the cache. `pack create` is idempotent on the smolvm side but
# takes several seconds even on a no-op rebuild.
_SMOLMACHINE_CACHE_DIR = Path.home() / ".cache" / "bot-bottle" / "smolmachines"


# Container-internal listening ports for each bundle daemon. The
# sidecar VM publishes each one on a random host loopback port, and
# the launch flow wraps those raw ports with per-bottle forwarders.
_GIT_HTTP_PORT = 9420
_SUPERVISE_PORT = SUPERVISE_PORT


@contextmanager
def launch(
    plan: SmolmachinesBottlePlan,
    *,
    provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
) -> Generator[SmolmachinesBottle, None, None]:
    """Build + run the bottle and yield a handle; tear everything
    down on exit. Errors during bringup unwind any partial state
    via the ExitStack."""
    stack = ExitStack()
    try:
        loopback_ip, network = _allocate_resources(plan, stack)
        plan = _mint_certs(plan)
        proxy_host = loopback_ip
        plan = _start_bundle(plan, network, proxy_host, stack)

        agent_from_path = _agent_from_path(plan)

        _launch_vm(plan, agent_from_path, proxy_host, stack)
        _init_vm(plan, proxy_host)

        bottle = SmolmachinesBottle(
            plan.machine_name,
            prompt_path=None,
            guest_env=plan.guest_env,
            agent_command=plan.agent_command,
            agent_prompt_mode=plan.agent_prompt_mode,
            agent_provider_template=plan.agent_provider_template,
            terminal_title=f"{plan.spec.label} ({plan.spec.agent_name})" if plan.spec.label else plan.spec.agent_name,
            terminal_color=plan.spec.color,
            agent_workdir=plan.workspace_plan.workdir,
        )
        bottle.prompt_path = provision(plan, bottle)

        yield bottle
    finally:
        _teardown_smolmachines(stack, plan)


def _teardown_smolmachines(
    stack: ExitStack,
    plan: SmolmachinesBottlePlan,
) -> None:
    """Unwind the ExitStack, then revoke any provisioned deploy keys.

    ExitStack errors are caught and logged (non-fatal) so that key
    revocation always runs. Revocation errors propagate — a stranded
    deploy key is a security concern the operator must address."""
    teardown_exc: BaseException | None = None
    try:
        stack.close()
    except BaseException as exc:  # noqa: W0718 — teardown must not fail
        teardown_exc = exc
        warn(f"smolmachines teardown failed: {exc!r}")
    bottle = plan.manifest.bottle
    revoke_git_gate_provisioned_keys(bottle, git_gate_state_dir(plan.slug))
    if teardown_exc is not None:
        raise teardown_exc


def _allocate_resources(
    plan: SmolmachinesBottlePlan,
    stack: ExitStack,
) -> tuple[str, str]:
    """Reserve a per-bottle host address.

    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)
    return loopback_ip, network


def _mint_certs(plan: SmolmachinesBottlePlan) -> SmolmachinesBottlePlan:
    """Mint the egress MITM CA and return the plan with CA paths filled."""
    egress_ca_host, egress_ca_cert_only = egress_tls_init(
        egress_state_dir(plan.slug),
    )
    egress_plan = dataclasses.replace(
        plan.egress_plan,
        mitmproxy_ca_host_path=egress_ca_host,
        mitmproxy_ca_cert_only_host_path=egress_ca_cert_only,
    )
    return dataclasses.replace(plan, egress_plan=egress_plan)


def _start_bundle(
    plan: SmolmachinesBottlePlan,
    network: str,
    proxy_host: str,
    stack: ExitStack,
) -> SmolmachinesBottlePlan:
    """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))
    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(
    plan: SmolmachinesBottlePlan,
) -> SmolmachinesBottlePlan:
    if not plan.git_gate_plan.upstreams:
        return plan
    git_gate_plan = provision_git_gate_dynamic_keys(
        plan.manifest.bottle,
        plan.git_gate_plan,
        git_gate_state_dir(plan.slug),
    )
    return dataclasses.replace(plan, git_gate_plan=git_gate_plan)


def _discover_urls(
    plan: SmolmachinesBottlePlan,
    proxy_host: str,
    published_ports: dict[int, int],
) -> SmolmachinesBottlePlan:
    """Stamp URLs + guest_env from per-bottle forwarder ports.

    `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 = 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 = 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 = 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")
    no_proxy = f"{existing_no_proxy},{proxy_host}"
    guest_env = {
        **plan.guest_env,
        "HTTPS_PROXY": agent_proxy_url,
        "HTTP_PROXY":  agent_proxy_url,
        "https_proxy": agent_proxy_url,
        "http_proxy":  agent_proxy_url,
        "NO_PROXY":    no_proxy,
        "no_proxy":    no_proxy,
    }
    if agent_git_gate_host:
        guest_env["GIT_GATE_URL"] = f"http://{agent_git_gate_host}"
    if agent_supervise_url:
        guest_env["MCP_SUPERVISE_URL"] = agent_supervise_url
    for entry in egress_agent_env_entries(plan.egress_plan):
        name, value = entry.split("=", 1)
        guest_env[name] = value

    return dataclasses.replace(
        plan,
        guest_env=guest_env,
        agent_proxy_url=agent_proxy_url,
        agent_git_gate_host=agent_git_gate_host,
        agent_supervise_url=agent_supervise_url,
    )


def _launch_vm(
    plan: SmolmachinesBottlePlan,
    agent_from_path: Path,
    proxy_host: str,
    stack: ExitStack,
) -> None:
    """Create, patch, and start the smolvm VM; register teardown.

    --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"
    # Destroy any leftover machine from a previous run that didn't
    # clean up (e.g. crash, interrupted teardown).
    try:
        _smolvm.machine_stop(plan.machine_name)
    except _smolvm.SmolvmError:
        pass
    try:
        _smolvm.machine_delete(plan.machine_name)
    except _smolvm.SmolvmError:
        pass
    _smolvm.machine_create(
        plan.machine_name,
        from_path=agent_from_path,
        allow_cidrs=[tsi_cidr],
        env=plan.guest_env,
    )
    stack.callback(_smolvm.machine_delete, plan.machine_name)
    # Confirm the booted VM's TSI allowlist will actually enforce the
    # /32 before start (smolvm 0.8.0 silently drops `--allow-cidr`
    # with `--from`, so the persisted state DB is patched if needed).
    # Fails closed if enforcement can't be confirmed.
    _loopback.force_allowlist(plan.machine_name, [tsi_cidr])
    _smolvm.machine_start(plan.machine_name)
    stack.callback(_smolvm.machine_stop, plan.machine_name)


def _init_vm(plan: SmolmachinesBottlePlan, proxy_host: str) -> None:
    """Repair filesystem ownership, enforce loopback isolation, and
    wait for exec channel readiness.

    Ownership repair: smolvm's pack process remaps files to the host
    invoker's uid (e.g. 501 on macOS, 1000 on Linux). The chowns use
    names not numbers so they're correct on either. /home/node must
    be node:node so Claude Code can write ~/.claude.json; /tmp +
    /var/tmp need root mode 1777 so non-root processes can create
    per-uid scratch dirs.

    Loopback isolation (Linux only): on macOS, smolvm's TSI allowlist
    restricts which host loopback IPs the guest can reach. On Linux,
    the allowlist DB patch crashes TSI boot, so we enforce the same
    /32 scope with guest-side iptables instead. The rules allow
    outbound to proxy_host/32, then reject all other 127.0.0.0/8.
    Since the agent runs as non-root (node), it cannot flush these
    rules.

    mkdir -p guards: when booting from a committed snapshot, /tmp and
    /var/tmp are excluded from the archive (they're ephemeral and their
    stale contents would have wrong uid after smolvm's uid remap). The
    directories must be created before chown/chmod can set permissions.

    wait_exec_ready polls until the exec channel is ready for the
    subsequent provision calls, replacing the empirical sleep."""
    _smolvm.machine_exec(plan.machine_name, [
        "sh", "-c",
        "mkdir -p /tmp /var/tmp && "
        "chown -R node:node /home/node && "
        "chown root:root /tmp /var/tmp && "
        "chmod 1777 /tmp /var/tmp",
    ])
    if not _loopback._is_macos():
        _enforce_loopback_isolation(plan.machine_name, proxy_host)
    _smolvm.wait_exec_ready(plan.machine_name)


def _enforce_loopback_isolation(machine_name: str, proxy_host: str) -> None:
    """Install iptables rules restricting the guest to proxy_host/32.

    On Linux, smolvm's TSI bridges the full host loopback into the
    guest, and the allow-cidr DB patch is incompatible with TSI.
    Guest-side iptables achieves the same per-bottle isolation:
    only the allocated loopback alias is reachable, so the agent
    can't probe other bottles' ports or other host services.

    Runs as root (exec default); the agent process runs as `node`
    (non-root) and cannot modify iptables rules."""
    result = _smolvm.machine_exec(machine_name, [
        "sh", "-c",
        # Allow the bottle's allocated loopback alias.
        f"iptables -A OUTPUT -d {proxy_host}/32 -j ACCEPT && "
        # Drop all other loopback destinations. DROP (not REJECT)
        # because the libkrun kernel lacks the REJECT target module.
        # Connections to blocked addresses time out rather than fail
        # fast, which is acceptable — the agent shouldn't be probing
        # non-allowed loopback addresses.
        "iptables -A OUTPUT -d 127.0.0.0/8 -j DROP",
    ])
    if result.returncode != 0:
        warn(
            f"guest iptables setup failed (exit {result.returncode}): "
            f"{(result.stderr or '').strip() or '<no stderr>'}. "
            f"Per-bottle loopback isolation is not enforced."
        )


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}"


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 _stage_sidecar_data(plan: SmolmachinesBottlePlan) -> Path:
    """Stage egress + git-gate files into one virtiofs-mountable dir.

    libkrun limits total mounts + port-mappings to 5. With 3 daemon
    ports the sidecar VM can carry at most 2 mounts (the second is
    the supervise DB). Egress and git-gate share a single mount:

        <staging>/egress/     → _EGRESS_CONFDIR_IN_VM
        <staging>/git-gate/   → _GIT_GATE_SCRIPTS_DIR_IN_VM

    The mount is writable so mitmproxy can write combined-trust.pem
    and cache per-host certs under the egress subdir."""
    staging = egress_state_dir(plan.slug) / "smolvm-sidecar-data"

    # --- egress subdir ---
    confdir = staging / "egress"
    confdir.mkdir(parents=True, exist_ok=True)
    ep = plan.egress_plan
    shutil.copy2(str(ep.mitmproxy_ca_host_path), str(confdir / "mitmproxy-ca.pem"))
    if ep.routes:
        shutil.copy2(str(ep.routes_path), str(confdir / "routes.yaml"))

    # --- git-gate subdir (only when upstreams are configured) ---
    gp = plan.git_gate_plan
    if gp.upstreams:
        scripts_dir = staging / "git-gate"
        scripts_dir.mkdir(parents=True, exist_ok=True)
        shutil.copy2(str(gp.entrypoint_script), str(scripts_dir / "entrypoint.sh"))
        shutil.copy2(str(gp.hook_script), str(scripts_dir / "pre-receive"))
        shutil.copy2(str(gp.access_hook_script), str(scripts_dir / "access-hook"))
        for name in ("entrypoint.sh", "pre-receive", "access-hook"):
            (scripts_dir / name).chmod(0o755)

        # Patch paths: the rendered entrypoint hardcodes /git-gate/creds/
        # and /etc/git-gate/ for hooks; rewrite both to the in-VM subdir.
        ep_path = scripts_dir / "entrypoint.sh"
        text = ep_path.read_text()
        text = text.replace("/git-gate/creds/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/creds/")
        text = text.replace("/etc/git-gate/", f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/")
        ep_path.write_text(text)

        creds_dir = scripts_dir / "creds"
        creds_dir.mkdir(exist_ok=True)
        for u in gp.upstreams:
            keypath = Path(expand_tilde(u.identity_file))
            dest_key = creds_dir / f"{u.name}-key"
            shutil.copy2(str(keypath), str(dest_key))
            dest_key.chmod(0o600)
            if u.known_hosts_file:
                dest_kh = creds_dir / f"{u.name}-known_hosts"
                shutil.copy2(str(u.known_hosts_file), str(dest_kh))
                dest_kh.chmod(0o600)

    return staging


def _bundle_launch_spec(
    plan: SmolmachinesBottlePlan, network: str, proxy_host: str,
) -> _bundle.BundleLaunchSpec:
    """Build a BundleLaunchSpec from the resolved inner Plans.

    Daemons in the CSV:
      - egress is always present.
      - git-gate + git-http are conditional on plan.git_gate_plan.upstreams.
      - supervise is conditional on plan.supervise_plan.

    Env + volumes are the union of the sidecar daemons' needs, with
    daemon-private values only (HTTPS_PROXY is scoped to the
    egress process by egress_entrypoint.sh — see PRD 0024's bundle
    bind-address PR)."""
    daemons: list[str] = ["egress"]
    env: list[str] = []
    volumes: list[tuple[str, str, bool]] = []

    # --- egress + git-gate (single mount) ---------------------
    # Stage both into one dir and mount it at _SIDECAR_DATA_DIR_IN_VM.
    # libkrun limits mounts + port-mappings to 5; with 3 daemon ports
    # we can carry at most 2 mounts (this one + supervise DB).
    # Writable so egress_entrypoint.sh can write combined-trust.pem
    # and mitmproxy can create its per-host cert cache.
    ep = plan.egress_plan
    gp = plan.git_gate_plan
    staging = _stage_sidecar_data(plan)
    volumes.append((str(staging), _SIDECAR_DATA_DIR_IN_VM, False))
    # Tell the egress entrypoint where to find its CA + routes.
    env.append(f"EGRESS_CONFDIR={_EGRESS_CONFDIR_IN_VM}")
    # Always set EGRESS_ROUTES so the addon reads from the confdir path
    # even when no routes were configured at launch (apply_routes_change
    # writes here and a SIGHUP causes the addon to pick them up).
    env.append(f"EGRESS_ROUTES={_EGRESS_CONFDIR_IN_VM}/routes.yaml")
    env.extend(egress_sidecar_env_entries(ep))

    if gp.upstreams:
        daemons += ["git-gate", "git-http"]

    # --- supervise --------------------------------------------
    sp = plan.supervise_plan
    if sp is not None:
        daemons.append("supervise")
        env += [
            f"SUPERVISE_BOTTLE_SLUG={plan.slug}",
            f"SUPERVISE_DB_PATH={DB_PATH_IN_CONTAINER}",
            f"SUPERVISE_PORT={SUPERVISE_PORT}",
        ]
        # virtiofs requires directory mount — mount the DB's parent
        # dir so bot-bottle.db lands at the right in-VM path.
        volumes.append((
            str(sp.db_path.parent),
            str(Path(DB_PATH_IN_CONTAINER).parent),
            False,
        ))

    # Container ports the agent reaches from the smolvm guest —
    # published on `proxy_host` so the TSI allowlist and the docker
    # port-forward bindings point at the same IP. Egress is always
    # the agent's HTTP/HTTPS proxy.
    ports_to_publish: list[int] = [_EGRESS_PORT]
    if gp.upstreams:
        ports_to_publish.append(_GIT_HTTP_PORT)
    if sp is not None:
        ports_to_publish.append(_SUPERVISE_PORT)

    return _bundle.BundleLaunchSpec(
        slug=plan.slug,
        network_name=network,
        subnet=plan.bundle_subnet,
        gateway=plan.bundle_gateway,
        bundle_ip=plan.bundle_ip,
        daemons_csv=",".join(daemons),
        environment=tuple(env),
        volumes=tuple(volumes),
        ports_to_publish=tuple(ports_to_publish),
        publish_host_ip=proxy_host,
    )


def _resolve_token_env(
    plan: SmolmachinesBottlePlan, host_env: dict[str, str],
) -> dict[str, str]:
    """Resolve the egress token env-var values from the host's
    environ so they reach the bundle's process env via docker's
    `-e NAME` inheritance. Empty when no routes declare auth."""
    effective_env = {**host_env, **plan.agent_provision.provisioned_env}
    return egress_resolve_token_values(plan.egress_plan.token_env_map, effective_env)


def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
    """Return the `.smolmachine` artifact used for `machine create --from`.

    Prefer a committed VM artifact when one is recorded and still
    present. If the file was removed, fall back to the normal image
    build + pack cache path.
    """
    committed = read_committed_image(plan.slug)
    if committed:
        committed_path = Path(committed)
        if committed_path.is_file():
            info(f"using committed smolmachine {str(committed_path)!r}")
            return committed_path

    # 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
    # dashboard's preflight modal.
    return _ensure_smolmachine(
        plan.agent_image,
        dockerfile=plan.agent_dockerfile_path,
    )


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
    `~/.cache/bot-bottle/smolmachines/` keyed by the docker image
    ID (so a Dockerfile change automatically invalidates the cache).

    Returns the `.smolmachine.smolmachine` sidecar path — that's
    the file `machine create --from` consumes (pack create produces
    a launcher binary at `.smolmachine` plus the sidecar alongside
    it; the sidecar is the actual artifact).

    Conversion path: `docker build` (the existing layer cache
    makes no-change rebuilds cheap) → `docker save` to a tarball
    → spin up an ephemeral registry on a private docker network →
    `crane push --insecure` from a one-shot container on the same
    network → `smolvm pack create --image localhost:<host port>/...`
    → tear down the registry + network. The crane push detour
    sidesteps the Docker-Desktop daemon's HTTPS preference for
    non-loopback registries — see the `local_registry` module
    docstring for the gory details.

    Each pack-create costs several seconds even on a hot cache,
    so we skip the whole pipeline when the cached sidecar is
    already on disk for this image ID."""
    _SMOLMACHINE_CACHE_DIR.mkdir(parents=True, exist_ok=True)
    docker_mod.build_image(image_ref, _REPO_DIR, dockerfile=dockerfile)
    # `sha256:abcd...` -> `abcd...` first 16 chars: short enough to
    # keep filenames manageable, long enough to make collisions
    # astronomically unlikely.
    digest = docker_mod.image_id(image_ref).split(":", 1)[-1][:16]
    binary = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine"
    sidecar = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
    if sidecar.is_file():
        return sidecar
    tarball = _SMOLMACHINE_CACHE_DIR / f"{digest}.image.tar"
    docker_mod.save(image_ref, str(tarball))
    # On Linux, `docker save -o` writes the tarball with owner-only
    # permissions (mode 600). The crane push container runs as UID
    # 65532 (distroless nonroot) and can't read it through a bind
    # mount unless world-read is set. The tarball is temporary and
    # lives in ~/.cache, so 644 is safe.
    tarball.chmod(0o644)
    try:
        with ephemeral_registry() as handle:
            push_ref = f"{handle.push_endpoint}/bot-bottle:{digest}"
            pack_ref = f"{handle.pull_endpoint}/bot-bottle:{digest}"
            crane_push_tarball(handle, str(tarball), push_ref)
            _smolvm.pack_create(pack_ref, binary)
    finally:
        # Tarball is ~500MB-1GB for the agent image; reclaim once
        # the smolmachine artifact exists. The artifact itself is
        # the long-lived cache entry.
        tarball.unlink(missing_ok=True)
    return sidecar