Compare commits

..

2 Commits

Author SHA1 Message Date
didericis dc9fdc8563 fix(smolmachines): enforce per-bottle loopback isolation on Linux via iptables
lint / lint (push) Successful in 2m7s
test / unit (pull_request) Failing after 55s
test / integration (pull_request) Successful in 25s
test / coverage (pull_request) Failing after 1m2s
On Linux, smolvm's TSI allowlist DB patch crashes boot, so
force_allowlist is a no-op. This left the agent VM with full
host loopback access — a security regression from the macOS
behavior.

Fix: install guest-side iptables rules in _init_vm that ACCEPT
traffic to the bottle's allocated loopback alias (proxy_host/32)
and DROP all other 127.0.0.0/8 destinations. The agent runs as
non-root (node) and cannot flush the rules.

- Add iptables to claude and codex agent Dockerfiles
- Use DROP (not REJECT) — libkrun kernel lacks the REJECT module
- Skip on macOS where force_allowlist handles it natively

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-10 17:11:27 -04:00
didericis d76ce89c87 fix(smolmachines): Linux compat for TSI networking and git-gate paths
lint / lint (push) Successful in 2m8s
test / unit (pull_request) Failing after 58s
test / integration (pull_request) Successful in 23s
test / coverage (pull_request) Failing after 1m1s
Three fixes for Linux smolvm support:

1. Skip force_allowlist DB patch on Linux — patching allowed_cidrs
   into smolvm's state DB crashes TSI boot on Linux. Per-bottle CIDR
   isolation is not enforced on Linux until smolvm fixes the
   --allow-cidr + TSI interaction (known limitation).

2. Patch /etc/git-gate/ references in the staged entrypoint to
   /bot-bottle-data/git-gate/ so git-daemon finds its hooks and
   access-hook at the new consolidated mount path.

3. Drop explicit --net-backend tsi — the default on this smolvm
   build is already TSI; explicitly requesting it is incompatible
   with -p port mappings on Linux.

Co-Authored-By: Claude Opus 4.6 <noreply@anthropic.com>
2026-07-10 17:01:23 -04:00
32 changed files with 201 additions and 1453 deletions
+4 -37
View File
@@ -36,10 +36,10 @@ import os
import shlex
import sys
from abc import ABC, abstractmethod
from contextlib import AbstractContextManager, contextmanager
from contextlib import AbstractContextManager
from dataclasses import dataclass
from pathlib import Path
from typing import Any, Generator, Generic, Sequence, TypeVar
from typing import Any, Generic, Sequence, TypeVar
from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
from ..egress import EgressPlan
@@ -78,9 +78,6 @@ class BottleSpec:
# True when launched via --headless (no TTY, no interactive prompts).
# The git-gate host-key preflight uses this to error rather than prompt.
headless: bool = False
# Image startup policy. "fresh" preserves the normal build path;
# "cached" reuses the current local image/artifact without rebuilding.
image_policy: str = "fresh"
@dataclass(frozen=True)
@@ -276,18 +273,6 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan)
@dataclass(frozen=True)
class BottleImages:
"""Resolved image references (or artifact paths) for a bottle launch.
For Docker/macOS-container backends, `agent` and `sidecar` are string
image refs. For the smolmachines backend they are Path objects pointing
to pre-built `.smolmachine` artifacts."""
agent: str | Path
sidecar: str | Path
class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Abstract base for selectable bottle backends. Concrete subclasses
(e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -447,27 +432,9 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
prompt file, Dockerfile path, and guest home all live on
`agent_provision_plan` — the source of truth."""
def prelaunch_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. Called by the CLI before
launch so the operator can be prompted outside the launch context."""
@contextmanager
def launch(self, plan: PlanT) -> Generator[Bottle, None, None]:
"""Template: build or load images, then delegate to _launch_impl."""
images = self._build_or_load_images(plan)
with self._launch_impl(plan, images) as bottle:
yield bottle
@abstractmethod
def _build_or_load_images(self, plan: PlanT) -> BottleImages:
"""Return the agent and sidecar image references (or artifact paths)
for this plan, building fresh images when the policy requires it."""
@abstractmethod
def _launch_impl(self, plan: PlanT, images: BottleImages) -> AbstractContextManager[Bottle]:
"""Bring up the bottle using pre-resolved images; yield a handle; tear down on exit."""
def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]:
"""Build/run the bottle and yield a handle; tear down on exit."""
def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
"""Copy host-side files (CA cert, prompt, skills, .git) into
+3 -9
View File
@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan
from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from .. import ActiveAgent, BottleBackend, BottleSpec
from . import cleanup as _cleanup
from . import enumerate as _enumerate
from . import launch as _launch
@@ -85,15 +85,9 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
stage_dir=stage_dir,
)
def prelaunch_checks(self, plan: DockerBottlePlan) -> None:
_launch.stale_checks(plan)
def _build_or_load_images(self, plan: DockerBottlePlan) -> BottleImages:
return _launch.build_or_load_images(plan)
@contextmanager
def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
with _launch.launch(plan, images, provision=self.provision) as bottle:
def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]:
with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle
def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:
+4 -5
View File
@@ -180,6 +180,10 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
service: dict[str, Any] = {
"image": SIDECAR_BUNDLE_IMAGE,
"build": {
"context": _REPO_DIR,
"dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
},
"container_name": sidecar_bundle_container_name(plan.slug),
"networks": {
"internal": {"aliases": internal_aliases},
@@ -188,11 +192,6 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
"environment": env,
"volumes": volumes,
}
if plan.spec.image_policy != "cached":
service["build"] = {
"context": _REPO_DIR,
"dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
}
return service
+19 -58
View File
@@ -41,9 +41,7 @@ from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...image_cache import check_stale
from ...log import die, info, warn
from .. import BottleImages
from ...log import info, warn
from . import network as network_mod
from . import util as docker_mod
from .bottle import DockerBottle
@@ -65,57 +63,22 @@ 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.
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
def build_or_load_images(plan: DockerBottlePlan) -> BottleImages:
"""Resolve the agent and sidecar image refs for this plan.
Returns the committed snapshot if one exists, the local cached images
when the policy is 'cached', or builds fresh images and returns those."""
committed = read_committed_image(plan.slug)
if committed and docker_mod.image_exists(committed):
info(f"using committed image {committed!r}")
return BottleImages(agent=committed, sidecar=SIDECAR_BUNDLE_IMAGE)
if plan.spec.image_policy == "cached":
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}")
return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
docker_mod.build_image(plan.image, _REPO_DIR, dockerfile=plan.dockerfile_path)
return BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
@contextmanager
def launch(
plan: DockerBottlePlan,
images: BottleImages,
*,
provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
) -> Generator[DockerBottle, None, None]:
"""Launch and provision a Docker bottle via compose. Teardown on exit."""
"""Build, launch, and provision a Docker bottle via compose.
Teardown on exit."""
stack = ExitStack()
# Stamp the resolved agent image ref into the plan so compose rendering
# picks up the right image (may be a committed snapshot or cached ref).
plan = dataclasses.replace(
plan,
agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
)
_bottle_for_revoke = plan.manifest.bottle
_git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
@@ -132,6 +95,22 @@ def launch(
)
try:
# Step 1: agent image. Use a committed snapshot when one exists
# and is present in the local daemon; otherwise build from the
# Dockerfile. Sidecar images get built lazily by `docker compose
# up` via the renderer's `build:` directives.
committed = read_committed_image(plan.slug)
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),
)
else:
docker_mod.build_image(
plan.image, _REPO_DIR,
dockerfile=plan.dockerfile_path,
)
internal_network = network_mod.network_name_for_slug(plan.slug)
egress_network = network_mod.network_egress_name_for_slug(plan.slug)
@@ -228,21 +207,3 @@ def launch(
yield bottle
finally:
teardown()
def stale_checks(plan: DockerBottlePlan) -> None:
"""Raise StaleImageError if a cached image is older than the configured
threshold. Only runs when image_policy is 'cached'. Called by the backend
class's _image_stale_checks before _launch_impl starts any resources."""
if plan.spec.image_policy != "cached":
return
committed = read_committed_image(plan.slug)
if committed and docker_mod.image_exists(committed):
check_stale(f"agent image {committed!r}", docker_mod.image_created_at(committed))
return
for label, ref in [
("agent image", plan.image),
("sidecar image", SIDECAR_BUNDLE_IMAGE),
]:
if docker_mod.image_exists(ref):
check_stale(f"{label} {ref!r}", docker_mod.image_created_at(ref))
-40
View File
@@ -4,7 +4,6 @@ existence, and building images."""
from __future__ import annotations
from datetime import datetime, timezone
import re
import shutil
import subprocess
@@ -187,45 +186,6 @@ def image_id(ref: str) -> str:
return r.stdout.strip()
def image_created_at(ref: str) -> datetime:
"""Return Docker's image Created timestamp as an aware UTC datetime."""
r = subprocess.run(
["docker", "image", "inspect", "--format", "{{.Created}}", ref],
capture_output=True,
text=True,
check=False,
)
if r.returncode != 0:
die(
f"docker image inspect for {ref!r} failed: "
f"{(r.stderr or '').strip() or '<no stderr>'}"
)
raw = r.stdout.strip()
try:
return _parse_docker_timestamp(raw)
except ValueError:
die(f"docker image inspect for {ref!r} returned invalid Created timestamp: {raw!r}")
def _parse_docker_timestamp(raw: str) -> datetime:
text = raw.strip()
if text.endswith("Z"):
text = text[:-1] + "+00:00"
dot = text.find(".")
if dot != -1:
tz_plus = text.find("+", dot)
tz_minus = text.find("-", dot)
tz_candidates = [pos for pos in (tz_plus, tz_minus) if pos != -1]
if tz_candidates:
tz_pos = min(tz_candidates)
frac = text[dot + 1:tz_pos]
text = text[:dot + 1] + frac[:6].ljust(6, "0") + text[tz_pos:]
dt = datetime.fromisoformat(text)
if dt.tzinfo is None:
dt = dt.replace(tzinfo=timezone.utc)
return dt.astimezone(timezone.utc)
def save(ref: str, output: str) -> None:
"""`docker save REF -o OUTPUT`. Writes a tarball of the image
layers + manifest to the host path. Used by smolmachines
+4 -10
View File
@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan
from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from .. import ActiveAgent, BottleBackend, BottleSpec
from . import cleanup as _cleanup
from . import enumerate as _enumerate
from . import launch as _launch
@@ -67,17 +67,11 @@ class MacosContainerBottleBackend(
stage_dir=stage_dir,
)
def prelaunch_checks(self, plan: MacosContainerBottlePlan) -> None:
_launch.stale_checks(plan)
def _build_or_load_images(self, plan: MacosContainerBottlePlan) -> BottleImages:
return _launch.build_or_load_images(plan)
@contextmanager
def _launch_impl(
self, plan: MacosContainerBottlePlan, images: BottleImages
def launch(
self, plan: MacosContainerBottlePlan
) -> Generator[MacosContainerBottle, None, None]:
with _launch.launch(plan, images, provision=self.provision) as bottle:
with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle
def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:
+22 -52
View File
@@ -32,9 +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 .. import BottleImages
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde
from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
@@ -73,51 +71,17 @@ def sidecar_container_name(slug: str) -> str:
return f"bot-bottle-sidecars-{slug}"
def build_or_load_images(plan: MacosContainerBottlePlan) -> BottleImages:
"""Resolve agent and sidecar image refs. Builds the sidecar when needed,
but never builds the agent when policy is 'cached'."""
committed = read_committed_image(plan.slug)
if committed and container_mod.image_exists(committed):
info(f"using committed image {committed!r}")
if plan.spec.image_policy != "cached":
container_mod.build_image(SIDECAR_BUNDLE_IMAGE, _REPO_DIR, dockerfile=SIDECAR_BUNDLE_DOCKERFILE)
return BottleImages(agent=committed, sidecar=SIDECAR_BUNDLE_IMAGE)
if plan.spec.image_policy == "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 BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
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 BottleImages(agent=plan.image, sidecar=SIDECAR_BUNDLE_IMAGE)
@contextmanager
def launch(
plan: MacosContainerBottlePlan,
images: BottleImages,
*,
provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
) -> Generator[MacosContainerBottle, None, None]:
"""Run, provision, and yield an Apple Container bottle."""
"""Build, run, provision, and yield an Apple Container bottle."""
stack = ExitStack()
bottle_for_revoke = plan.manifest.bottle
git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
plan = dataclasses.replace(
plan,
agent_provision=dataclasses.replace(plan.agent_provision, image=str(images.agent)),
)
def teardown() -> None:
teardown_exc: BaseException | None = None
try:
@@ -131,6 +95,7 @@ def launch(
try:
plan = _mint_certs(plan)
plan = _build_images(plan)
internal_network = internal_network_name(plan.slug)
egress_network = egress_network_name(plan.slug)
@@ -183,23 +148,28 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
return dataclasses.replace(plan, egress_plan=egress_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
def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
container_mod.build_image(
SIDECAR_BUNDLE_IMAGE,
_REPO_DIR,
dockerfile=SIDECAR_BUNDLE_DOCKERFILE,
)
committed = read_committed_image(plan.slug)
if committed and container_mod.image_exists(committed):
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))
info(f"using committed image {committed!r}")
return dataclasses.replace(
plan,
agent_provision=dataclasses.replace(
plan.agent_provision,
image=committed,
),
)
container_mod.build_image(
plan.image,
_REPO_DIR,
dockerfile=plan.dockerfile_path,
)
return plan
def _create_networks(
@@ -10,7 +10,6 @@ import shutil
import subprocess
import tempfile
import time
from datetime import datetime, timezone
from typing import Iterable
from ...log import die, info
@@ -459,39 +458,6 @@ def image_id(ref: str) -> str:
raise AssertionError("unreachable")
def image_created_at(ref: str) -> datetime:
"""Return the image creation timestamp as an aware UTC datetime.
Parses the `created` field from `container image inspect` JSON output."""
result = subprocess.run(
[_CONTAINER, "image", "inspect", ref],
capture_output=True,
text=True,
check=False,
)
if result.returncode != 0:
die(
f"container image inspect for {ref!r} failed: "
f"{(result.stderr or '').strip() or '<no stderr>'}"
)
try:
data = json.loads(result.stdout or "{}")
except json.JSONDecodeError as exc:
die(f"container image inspect for {ref!r} returned malformed JSON: {exc}")
if isinstance(data, list) and data:
data = data[0]
if isinstance(data, dict):
value = data.get("created") or data.get("Created")
if isinstance(value, str) and value:
try:
ts = value.rstrip("Z")
return datetime.fromisoformat(ts).replace(tzinfo=timezone.utc)
except ValueError:
pass
die(f"container image inspect for {ref!r} did not include a creation timestamp")
raise AssertionError("unreachable")
def save(ref: str, output: str) -> None:
subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True)
+4 -10
View File
@@ -19,7 +19,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan
from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from .. import ActiveAgent, BottleBackend, BottleSpec
from . import cleanup as _cleanup
from . import enumerate as _enumerate
from . import launch as _launch
@@ -77,17 +77,11 @@ class SmolmachinesBottleBackend(
stage_dir=stage_dir,
)
def prelaunch_checks(self, plan: SmolmachinesBottlePlan) -> None:
_launch.stale_checks(plan)
def _build_or_load_images(self, plan: SmolmachinesBottlePlan) -> BottleImages:
return _launch.build_or_load_images(plan)
@contextmanager
def _launch_impl(
self, plan: SmolmachinesBottlePlan, images: BottleImages
def launch(
self, plan: SmolmachinesBottlePlan
) -> Generator[SmolmachinesBottle, None, None]:
with _launch.launch(plan, images, provision=self.provision) as bottle:
with _launch.launch(plan, provision=self.provision) as bottle:
yield bottle
def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str:
+67 -103
View File
@@ -35,14 +35,12 @@ from ...git_gate import (
provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys,
)
from ...image_cache import check_stale_path
from ...log import die, info, warn
from ...log import info, warn
from ...bottle_state import (
egress_state_dir,
git_gate_state_dir,
read_committed_image,
)
from .. import BottleImages
from . import loopback_alias as _loopback
from . import port_forward as _forward
from . import sidecar_bundle as _bundle
@@ -82,32 +80,13 @@ _GIT_HTTP_PORT = 9420
_SUPERVISE_PORT = SUPERVISE_PORT
def build_or_load_images(plan: SmolmachinesBottlePlan) -> BottleImages:
"""Return pre-built or freshly built smolmachine artifact paths."""
return BottleImages(
agent=_agent_from_path(plan),
sidecar=_sidecar_from_path(plan),
)
def _sidecar_from_path(plan: SmolmachinesBottlePlan) -> Path:
"""Return the sidecar bundle artifact path, building it if needed."""
if _image_policy(plan) == "cached":
return _cached_smolmachine(_bundle.SIDECAR_BUNDLE_IMAGE, label="sidecar")
return _ensure_smolmachine(
_bundle.SIDECAR_BUNDLE_IMAGE,
dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
)
@contextmanager
def launch(
plan: SmolmachinesBottlePlan,
images: BottleImages,
*,
provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
) -> Generator[SmolmachinesBottle, None, None]:
"""Run the bottle from pre-built images and yield a handle; tear everything
"""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()
@@ -115,10 +94,12 @@ def launch(
loopback_ip, network = _allocate_resources(plan, stack)
plan = _mint_certs(plan)
proxy_host = loopback_ip
plan = _start_bundle(plan, network, proxy_host, Path(images.sidecar), stack)
plan = _start_bundle(plan, network, proxy_host, stack)
_launch_vm(plan, Path(images.agent), proxy_host, stack)
_init_vm(plan)
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,
@@ -194,18 +175,21 @@ def _start_bundle(
plan: SmolmachinesBottlePlan,
network: str,
proxy_host: str,
sidecar_artifact: Path,
stack: ExitStack,
) -> SmolmachinesBottlePlan:
"""Build the BundleLaunchSpec, start the sidecar VM from the pre-resolved
artifact, wrap its raw smolVM-published loopback ports with per-bottle
forwarders, stamp agent URLs from those forwarder ports, 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))
artifact = _ensure_smolmachine(
bundle_spec.image,
dockerfile=_bundle.SIDECAR_BUNDLE_DOCKERFILE,
)
launch = _bundle.start_bundle_vm(
bundle_spec,
from_path=sidecar_artifact,
from_path=artifact,
host_env={**os.environ, **token_env},
)
stack.callback(_bundle.stop_bundle_vm, plan.slug)
@@ -338,17 +322,24 @@ def _launch_vm(
stack.callback(_smolvm.machine_stop, plan.machine_name)
def _init_vm(plan: SmolmachinesBottlePlan) -> None:
"""Repair filesystem ownership and wait for exec channel readiness.
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.
All folded into one sh -c to avoid back-to-back exec calls
immediately after machine_start (libkrun exec-channel race).
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
@@ -364,9 +355,41 @@ def _init_vm(plan: SmolmachinesBottlePlan) -> None:
"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"
@@ -422,15 +445,13 @@ def _stage_sidecar_data(plan: SmolmachinesBottlePlan) -> Path:
for name in ("entrypoint.sh", "pre-receive", "access-hook"):
(scripts_dir / name).chmod(0o755)
# Patch credential paths: the rendered entrypoint hardcodes
# /git-gate/creds/; rewrite to the in-VM git-gate subdir.
# 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"
ep_path.write_text(
ep_path.read_text().replace(
"/git-gate/creds/",
f"{_GIT_GATE_SCRIPTS_DIR_IN_VM}/creds/",
)
)
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)
@@ -551,9 +572,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
info(f"using committed smolmachine {str(committed_path)!r}")
return committed_path
if _image_policy(plan) == "cached":
return _cached_smolmachine(plan.agent_image, label="agent")
# Build the agent image and pack it into a `.smolmachine`
# artifact (or hit the per-Dockerfile-digest cache). Runs here,
# not in prepare, so the docker-build output doesn't garble the
@@ -564,60 +582,6 @@ def _agent_from_path(plan: SmolmachinesBottlePlan) -> Path:
)
def stale_checks(plan: SmolmachinesBottlePlan) -> None:
"""Raise StaleImageError if a cached smolmachine artifact is older than the
configured threshold. Only runs when image_policy is 'cached'. Checks the
committed agent artifact first, then the cache-keyed agent and sidecar
artifacts. Called by the backend class's _image_stale_checks before any
resources are allocated."""
if _image_policy(plan) != "cached":
return
committed = read_committed_image(plan.slug)
if committed:
committed_path = Path(committed)
if committed_path.is_file():
check_stale_path("agent smolmachine artifact", committed_path)
elif docker_mod.image_exists(plan.agent_image):
digest = docker_mod.image_id(plan.agent_image).split(":", 1)[-1][:16]
artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
if artifact.is_file():
check_stale_path("agent smolmachine artifact", artifact)
sidecar_image = _bundle.SIDECAR_BUNDLE_IMAGE
if docker_mod.image_exists(sidecar_image):
digest = docker_mod.image_id(sidecar_image).split(":", 1)[-1][:16]
sidecar_artifact = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
if sidecar_artifact.is_file():
check_stale_path("sidecar smolmachine artifact", sidecar_artifact)
def _image_policy(plan: object) -> str:
spec = getattr(plan, "spec", None)
return str(getattr(spec, "image_policy", "fresh"))
def _cached_smolmachine(image_ref: str, *, label: str) -> Path:
"""Return the cached smolmachine artifact for the current local image.
This is intentionally buildless: it inspects the existing local Docker
image ID and looks for the artifact keyed by that ID. If either side is
missing, the caller must use the fresh path to build/pack it.
"""
if not docker_mod.image_exists(image_ref):
die(
f"cached {label} image {image_ref!r} not found; "
"run without --cached-images to build it"
)
digest = docker_mod.image_id(image_ref).split(":", 1)[-1][:16]
sidecar = _SMOLMACHINE_CACHE_DIR / f"{digest}.smolmachine.smolmachine"
if not sidecar.is_file():
die(
f"cached {label} smolmachine artifact for {image_ref!r} not found; "
"run without --cached-images to build it"
)
info(f"using cached {label} smolmachine {str(sidecar)!r}")
return sidecar
def _ensure_smolmachine(image_ref: str, *, dockerfile: str = "") -> Path:
"""Build the agent docker image and convert it into a
`.smolmachine` artifact, caching the result under
@@ -152,23 +152,31 @@ def force_allowlist(machine_name: str, allowed_cidrs: list[str]) -> None:
"""Ensure the machine's persisted TSI allowlist equals
`allowed_cidrs`, failing **closed** if that can't be confirmed.
Runs on both macOS and Linux. It exists because smolvm 0.8.0
silently drops `--allow-cidr` when combined with `--from`, so
the allowlist has to be written into smolvm's persistent state
DB before `machine start`. Rather than assume the flag was
dropped, we read the persisted row and only patch when it
doesn't already match — so a newer smolvm that honors the flag
is left untouched.
macOS only. On Linux, smolvm's TSI defaults to full loopback
access and patching `allowed_cidrs` into the state DB crashes
TSI boot (the VM fails with "boot process exited (code 1)").
Per-bottle CIDR isolation is therefore not enforced on Linux
until smolvm fixes the `--allow-cidr` + TSI interaction. This
is a known limitation — the agent VM can reach all of host
loopback, not just its bottle's forwarder ports.
On macOS, smolvm 0.8.0 silently drops `--allow-cidr` when
combined with `--from`, so the allowlist has to be written
into smolvm's persistent state DB before `machine start`.
Rather than assume the flag was dropped, we read the persisted
row and only patch when it doesn't already match — so a newer
smolvm that honors the flag is left untouched.
Must run AFTER `smolvm machine create` (the row has to exist)
and BEFORE `smolvm machine start` (smolvm reads the row on
start; in-flight VMs don't pick up changes).
Fail-closed: if the state DB is missing, the row is missing, or
the allowlist still doesn't match after patching, we `die()`
rather than boot a VM whose egress confinement we can't verify
— an unconfirmed allowlist is a sandbox-escape risk (the agent
VM could reach all of host loopback)."""
Fail-closed (macOS): if the state DB is missing, the row is
missing, or the allowlist still doesn't match after patching,
we `die()` rather than boot a VM whose egress confinement we
can't verify."""
if not _is_macos():
return
want = list(allowed_cidrs)
if not _SMOLVM_DB_PATH.is_file():
die(
+4 -33
View File
@@ -36,7 +36,6 @@ from ..bottle_state import (
is_preserved,
mark_preserved,
)
from ..image_cache import StaleImageError
from ..log import info, die
from ..manifest import Manifest, ManifestIndex
from ._common import PROG, USER_CWD, read_tty_line
@@ -64,14 +63,6 @@ def cmd_start(argv: list[str]) -> int:
"skip all prompts. For orchestrators, CI, and webhooks."
),
)
parser.add_argument(
"--cached-images",
action="store_true",
help=(
"quickstart with existing local agent and sidecar images; "
"only valid with --headless"
),
)
parser.add_argument(
"--bottle",
action="append",
@@ -104,8 +95,6 @@ def cmd_start(argv: list[str]) -> int:
help="agent name defined in bot-bottle.json (omit to pick interactively)",
)
args = parser.parse_args(argv)
if args.cached_images and not args.headless:
die("--cached-images is only supported with --headless")
dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
@@ -153,10 +142,6 @@ def cmd_start(argv: list[str]) -> int:
label, color = tui.name_color_modal(default_label=agent_name)
label, color = _resolve_unique_label(label, color)
image_policy = _select_image_policy()
if image_policy is None:
return 0
spec = BottleSpec(
manifest=manifest,
agent_name=agent_name,
@@ -165,7 +150,6 @@ def cmd_start(argv: list[str]) -> int:
label=label,
color=color,
bottle_names=bottle_names,
image_policy=image_policy,
)
return _launch_bottle(
spec,
@@ -226,7 +210,6 @@ def _start_headless(
color=args.color or "",
bottle_names=bottle_names,
headless=True,
image_policy="cached" if args.cached_images else "fresh",
)
return _launch_bottle(
spec,
@@ -407,13 +390,6 @@ def _text_prompt_yes() -> bool:
return reply in ("y", "Y", "yes", "YES")
def _select_image_policy() -> str | None:
return tui.filter_select(
["fresh", "cached"],
title="Select image startup mode",
)
def _text_render_preflight():
def _render(plan: DockerBottlePlan, backend_name: str) -> None:
print(file=sys.stderr)
@@ -556,15 +532,6 @@ def _launch_bottle(
return 0
backend = get_bottle_backend(backend_name)
try:
backend.prelaunch_checks(plan)
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"):
return 0
with backend.launch(plan) as bottle:
agent_provider_template = getattr(plan, "agent_provider_template", "claude")
extra_args: tuple[str, ...] = ()
@@ -583,6 +550,10 @@ 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)
return 0
-71
View File
@@ -1,71 +0,0 @@
"""SQLite-backed bot-bottle configuration store."""
from __future__ import annotations
from pathlib import Path
try:
from .db_store import DbStore
from .migrations import TableMigrations
from .supervise_types import host_db_path
except ImportError:
from db_store import DbStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from supervise_types import host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS = 1
class ConfigStore(DbStore):
"""SQLite configuration for host-side bot-bottle settings."""
def __init__(self, db_path: Path | None = None) -> None:
migrations = TableMigrations("config_store", [
# v1 — host-side bot-bottle settings
"""
CREATE TABLE IF NOT EXISTS bot_bottle_config (
id INTEGER PRIMARY KEY CHECK (id = 1),
cached_image_stale_warning_days INTEGER NOT NULL DEFAULT 1
)
""",
])
super().__init__(db_path or host_db_path(), migrations)
def cached_image_stale_warning_days(self) -> int:
if not self.db_path.is_file():
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
with self._connect() as conn:
row = conn.execute(
"""
SELECT cached_image_stale_warning_days
FROM bot_bottle_config
WHERE id = 1
""",
).fetchone()
if row is None:
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
try:
return int(row["cached_image_stale_warning_days"])
except (TypeError, ValueError):
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
def set_cached_image_stale_warning_days(self, days: int) -> Path:
with self._connect() as conn:
conn.execute(
"""
INSERT INTO bot_bottle_config (id, cached_image_stale_warning_days)
VALUES (1, ?)
ON CONFLICT(id) DO UPDATE SET
cached_image_stale_warning_days = excluded.cached_image_stale_warning_days
""",
(days,),
)
self._chmod()
return self.db_path
__all__ = [
"DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS",
"ConfigStore",
]
+1 -1
View File
@@ -21,7 +21,7 @@ FROM node:22-slim
# to it) works against egress's bumped TLS without the agent needing
# local DNS.
RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils \
&& apt-get install -y --no-install-recommends git ca-certificates curl ripgrep iproute2 dnsutils iptables \
&& rm -rf /var/lib/apt/lists/*
# App-specific deps. Python isn't required by claude-code itself
+1 -1
View File
@@ -6,7 +6,7 @@
FROM node:22-slim
RUN apt-get update \
&& apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep \
&& apt-get install -y --no-install-recommends git ca-certificates curl procps ripgrep iptables \
&& rm -rf /var/lib/apt/lists/*
# App-specific deps. Python isn't required by codex itself
-42
View File
@@ -1,42 +0,0 @@
"""Shared helpers for cached-image quickstart stale checks."""
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
try:
from .config_store import ConfigStore
except ImportError:
from config_store import ConfigStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
class StaleImageError(Exception):
"""Raised when a cached image or artifact exceeds the configured staleness
threshold. Callers can catch this to prompt interactively; headless paths
let it propagate as a fatal error."""
def check_stale(label: str, created_at: datetime) -> None:
"""Raise StaleImageError if `created_at` is older than the configured
stale-warning threshold. Negative threshold disables the check."""
threshold_days = ConfigStore().cached_image_stale_warning_days()
if threshold_days < 0:
return
now = datetime.now(timezone.utc)
created = created_at.astimezone(timezone.utc)
age = now - created
if age.total_seconds() <= threshold_days * 86400:
return
raise StaleImageError(
f"cached {label} is {age.days} day(s) old; "
"quickstart does not verify it matches the current Dockerfile/context"
)
def check_stale_path(label: str, path: Path) -> None:
"""Raise StaleImageError if `path`'s mtime exceeds the staleness threshold."""
check_stale(label, datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc))
__all__ = ["StaleImageError", "check_stale", "check_stale_path"]
-4
View File
@@ -6,11 +6,9 @@ from pathlib import Path
try:
from .audit_store import AuditStore
from .config_store import ConfigStore
from .queue_store import QueueStore
except ImportError:
from audit_store import AuditStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from config_store import ConfigStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from queue_store import QueueStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
_instance: StoreManager | None = None
@@ -52,13 +50,11 @@ class StoreManager:
return (
QueueStore("", self.db_path).is_migrated()
and AuditStore(self.db_path).is_migrated()
and ConfigStore(self.db_path).is_migrated()
)
def migrate(self) -> None:
QueueStore("", self.db_path).migrate()
AuditStore(self.db_path).migrate()
ConfigStore(self.db_path).migrate()
__all__ = ["StoreManager"]
-12
View File
@@ -184,18 +184,6 @@ class TestCmdStartHeadless(unittest.TestCase):
)
self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"])
def test_cached_images_sets_cached_policy(self):
start_mod.cmd_start(
["--headless", "--cached-images", "researcher", "--bottle", "claude",
"--prompt", "Do it"]
)
self.assertEqual("cached", self._spec().image_policy)
def test_cached_images_requires_headless(self):
with self.assertRaises(Die):
start_mod.cmd_start(["--cached-images", "researcher"])
self._launch_mock.assert_not_called()
class TestPrepareWithPreflight(unittest.TestCase):
"""prepare_with_preflight calls render_preflight with the plan and backend name."""
-21
View File
@@ -57,12 +57,6 @@ class TestCmdStartSelector(unittest.TestCase):
self._bottle_picker_mock = self._bottle_picker_patch.start()
self._bottle_picker_mock.return_value = ["claude"] # default: one bottle selected
self._image_policy_patch = patch(
"bot_bottle.cli.start._select_image_policy",
return_value="fresh",
)
self._image_policy_patch.start()
self._env_patch = patch.dict(os.environ, {}, clear=False)
self._env_patch.start()
os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -72,7 +66,6 @@ class TestCmdStartSelector(unittest.TestCase):
self._launch_patch.stop()
self._agent_picker_patch.stop()
self._bottle_picker_patch.stop()
self._image_policy_patch.stop()
self._env_patch.stop()
# ------------------------------------------------------------------
@@ -131,19 +124,6 @@ class TestCmdStartSelector(unittest.TestCase):
spec = self._launch_mock.call_args[0][0]
self.assertEqual(("claude", "dev"), spec.bottle_names)
def test_image_policy_forwarded_to_spec(self):
with patch("bot_bottle.cli.start._select_image_policy", return_value="cached"):
start_mod.cmd_start(["researcher"])
self._launch_mock.assert_called_once()
spec = self._launch_mock.call_args[0][0]
self.assertEqual("cached", spec.image_policy)
def test_image_policy_cancel_returns_0(self):
with patch("bot_bottle.cli.start._select_image_policy", return_value=None):
rc = start_mod.cmd_start(["researcher"])
self.assertEqual(0, rc)
self._launch_mock.assert_not_called()
def test_empty_bottle_selection_forwarded(self):
self._bottle_picker_mock.return_value = []
start_mod.cmd_start(["researcher"])
@@ -235,7 +215,6 @@ class TestCmdStartLabelCollision(unittest.TestCase):
).start()
# Stub the bottle picker to always return a selection.
patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start()
patch("bot_bottle.cli.start._select_image_policy", return_value="fresh").start()
self.addCleanup(patch.stopall)
def test_no_collision_proceeds_without_reprompt(self):
-146
View File
@@ -1,146 +0,0 @@
"""Unit: _launch_bottle StaleImageError handling.
Exercises prelaunch_checks / backend.launch flow:
- headless mode die on stale
- interactive mode, user declines stop without launching
- interactive mode, user confirms skip stale check and launch once
"""
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 _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 from prelaunch_checks must call die()."""
import bot_bottle.cli.start as start_mod
backend_mock = MagicMock()
backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
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)
backend_mock.launch.assert_not_called()
def test_interactive_user_declines_stops_before_launch(self) -> None:
"""Interactive user answering 'n' → launch is never called."""
import bot_bottle.cli.start as start_mod
backend_mock = MagicMock()
backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
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)
backend_mock.launch.assert_not_called()
def test_interactive_user_confirms_launches_once(self) -> None:
"""Interactive user answering 'y' → prelaunch stale error is bypassed; launch called once."""
import bot_bottle.cli.start as start_mod
bottle_mock = MagicMock()
bottle_mock.name = "dev-abc"
backend_mock = MagicMock()
backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
backend_mock.launch.return_value = _ok_cm(bottle_mock)
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)
backend_mock.prelaunch_checks.assert_called_once()
backend_mock.launch.assert_called_once()
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"
backend_mock = MagicMock()
backend_mock.prelaunch_checks.side_effect = StaleImageError("image is 5 day(s) old")
backend_mock.launch.return_value = _ok_cm(bottle_mock)
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)
backend_mock.launch.assert_called_once()
if __name__ == "__main__":
unittest.main()
+1 -9
View File
@@ -118,7 +118,6 @@ def _plan(
with_egress: bool = False,
supervise: bool = False,
canary: bool = False,
image_policy: str = "fresh",
) -> DockerBottlePlan:
"""Build a fully-resolved DockerBottlePlan. Toggles cover the
matrix the renderer's conditional-service logic branches on."""
@@ -149,7 +148,6 @@ def _plan(
agent_name="demo",
copy_cwd=False,
user_cwd="/tmp/x",
image_policy=image_policy,
)
return DockerBottlePlan(
spec=spec,
@@ -267,8 +265,7 @@ class TestAgentAlwaysPresent(unittest.TestCase):
def test_agent_depends_only_on_sidecars(self):
# Bundle shape: the init supervisor owns intra-bundle daemon
# ordering, so the agent waits on the bundle container alone.
cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
for kwargs in cases:
for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]:
with self.subTest(**kwargs):
s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
self.assertEqual(["sidecars"], s["depends_on"])
@@ -303,11 +300,6 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"])
def test_cached_policy_omits_bundle_build(self):
sc = self._render(image_policy="cached")["services"]["sidecars"]
self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
self.assertNotIn("build", sc)
def test_bundle_container_name_uses_sidecars_prefix(self):
sc = self._render()["services"]["sidecars"]
self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])
-86
View File
@@ -1,86 +0,0 @@
"""Unit tests for the host-side configuration store."""
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from bot_bottle.config_store import (
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
ConfigStore,
)
from bot_bottle.store_manager import StoreManager
class TestConfigStore(unittest.TestCase):
def test_cached_image_warning_days_defaults_to_one(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
def test_cached_image_warning_days_reads_value(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
store.set_cached_image_stale_warning_days(7)
self.assertEqual(7, store.cached_image_stale_warning_days())
def test_config_schema_uses_explicit_settings_columns(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
with sqlite3.connect(store.db_path) as conn:
conn.row_factory = sqlite3.Row
columns = [
row["name"]
for row in conn.execute("PRAGMA table_info(bot_bottle_config)")
]
self.assertEqual([
"id",
"cached_image_stale_warning_days",
], columns)
def test_store_manager_includes_config_store(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
db = Path(tmp) / "bot-bottle.db"
manager = StoreManager(db)
self.assertFalse(manager.is_migrated())
manager.migrate()
self.assertTrue(manager.is_migrated())
def test_cached_image_warning_days_returns_default_when_db_missing(self) -> None:
# When the db file doesn't exist yet (parent exists, file doesn't),
# the store returns the default without touching the file.
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "missing.db")
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
def test_cached_image_warning_days_returns_default_on_null_value(self) -> None:
# If the row exists but the value is NULL (or not castable to int),
# the store falls back to the default.
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
db_path = Path(tmp) / "bot-bottle.db"
store = ConfigStore(db_path)
store.migrate()
# Write a NULL value directly.
with sqlite3.connect(db_path) as conn:
conn.execute(
"UPDATE bot_bottle_config SET cached_image_stale_warning_days = NULL WHERE id = 1"
)
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
if __name__ == "__main__":
unittest.main()
@@ -3,7 +3,6 @@
from __future__ import annotations
import contextlib
import dataclasses
import io
import tempfile
import unittest
@@ -17,7 +16,6 @@ from bot_bottle.backend.docker import launch as launch_mod
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestIndex
@@ -90,7 +88,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
committed_tag: str | None = None,
image_present: bool = True,
) -> list[str]:
"""Drive build_or_load_images() + launch() with the committed-image
"""Drive launch() through its full sequence with the committed-image
behaviour controlled by the arguments. Returns the images that were
passed to `build_image` (empty list if it was never called)."""
built: list[str] = []
@@ -124,9 +122,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()):
images = launch_mod.build_or_load_images(plan)
provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, images, provision=provision):
with launch_mod.launch(plan, provision=provision):
pass
return built
@@ -171,9 +168,8 @@ class TestLaunchCommittedImage(unittest.TestCase):
mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()):
images = launch_mod.build_or_load_images(plan)
provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, images, provision=provision):
with launch_mod.launch(plan, provision=provision):
pass
self.assertEqual(1, len(captured_plans))
@@ -184,24 +180,6 @@ class TestLaunchCommittedImage(unittest.TestCase):
built = self._run_launch(plan, committed_tag=None)
self.assertEqual([_DEFAULT_IMAGE], built)
def test_cached_images_skip_build_when_present(self) -> None:
base = _plan(self._tmp)
plan = dataclasses.replace(
base,
spec=dataclasses.replace(base.spec, image_policy="cached"),
)
built = self._run_launch(plan, committed_tag=None, image_present=True)
self.assertEqual([], built)
def test_cached_images_die_when_agent_missing(self) -> None:
base = _plan(self._tmp)
plan = dataclasses.replace(
base,
spec=dataclasses.replace(base.spec, image_policy="cached"),
)
with self.assertRaises(Die):
self._run_launch(plan, committed_tag=None, image_present=False)
def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
plan = _plan(self._tmp)
built = self._run_launch(
+2 -4
View File
@@ -16,7 +16,7 @@ from pathlib import Path
from unittest import mock
from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import BottleImages, BottleSpec
from bot_bottle.backend import BottleSpec
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
@@ -86,8 +86,6 @@ class TestTeardownWarning(unittest.TestCase):
plan = _plan(self._tmp)
buf = io.StringIO()
images = BottleImages(agent="bot-bottle-claude:latest", sidecar="bot-bottle-sidecars:latest")
with mock.patch.object(launch_mod.docker_mod, "build_image"), \
mock.patch.object(
launch_mod, "egress_tls_init",
@@ -117,7 +115,7 @@ class TestTeardownWarning(unittest.TestCase):
), \
contextlib.redirect_stderr(buf):
provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, images, provision=provision):
with launch_mod.launch(plan, provision=provision):
pass
output = buf.getvalue()
-46
View File
@@ -9,7 +9,6 @@ from __future__ import annotations
import subprocess
import unittest
from datetime import timezone
from unittest.mock import patch
from bot_bottle.backend.docker import util as docker_mod
@@ -55,51 +54,6 @@ class TestImageId(unittest.TestCase):
self.assertIn("missing:tag", die.call_args.args[0])
class TestImageCreatedAt(unittest.TestCase):
def test_parses_docker_timestamp_with_nanoseconds(self):
with patch.object(
docker_mod.subprocess, "run",
return_value=_ok(stdout="2026-07-06T15:33:47.123456789Z\n"),
) as run:
created = docker_mod.image_created_at("bot-bottle-claude:latest")
self.assertEqual(2026, created.year)
self.assertEqual(123456, created.microsecond)
self.assertEqual(timezone.utc, created.tzinfo)
self.assertEqual(
["docker", "image", "inspect", "--format", "{{.Created}}", "bot-bottle-claude:latest"],
run.call_args.args[0],
)
def test_dies_on_inspect_failure(self):
with patch.object(
docker_mod.subprocess, "run", return_value=_fail("No such image"),
), patch.object(
docker_mod, "die", side_effect=SystemExit("die"),
) as die:
with self.assertRaises(SystemExit):
docker_mod.image_created_at("missing:tag")
die.assert_called_once()
self.assertIn("missing:tag", die.call_args.args[0])
def test_dies_on_invalid_timestamp(self):
with patch.object(
docker_mod.subprocess, "run",
return_value=_ok(stdout="not-a-timestamp\n"),
), patch.object(
docker_mod, "die", side_effect=SystemExit("die"),
) as die:
with self.assertRaises(SystemExit):
docker_mod.image_created_at("some:tag")
die.assert_called_once()
self.assertIn("some:tag", die.call_args.args[0])
def test_parse_docker_timestamp_no_tzinfo_defaults_to_utc(self):
# A bare datetime with no tz offset should be treated as UTC.
dt = docker_mod._parse_docker_timestamp("2024-05-01T10:00:00.000000")
self.assertIsNotNone(dt.tzinfo)
self.assertEqual(timezone.utc, dt.tzinfo)
class TestSave(unittest.TestCase):
def test_save_runs_docker_save(self):
with patch.object(
-81
View File
@@ -1,81 +0,0 @@
"""Unit: image_cache.py — check_stale / check_stale_path."""
from __future__ import annotations
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path
class TestCheckStale(unittest.TestCase):
def _run(self, threshold: int, age_days: float) -> None:
created = datetime.now(tz=timezone.utc) - timedelta(days=age_days)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = threshold
check_stale("test image", created)
def test_negative_threshold_never_raises(self):
# Threshold < 0 means the check is disabled — always passes.
self._run(threshold=-1, age_days=9999)
def test_zero_threshold_raises_immediately(self):
# threshold=0 means any image is stale the moment it exists.
with self.assertRaises(StaleImageError):
self._run(threshold=0, age_days=0.1)
def test_within_threshold_does_not_raise(self):
# Age well under threshold — should pass silently.
self._run(threshold=7, age_days=2)
def test_at_threshold_does_not_raise(self):
# Exactly at the boundary is fine (<=, not <).
self._run(threshold=1, age_days=0.9999)
def test_exceeds_threshold_raises(self):
created = datetime.now(tz=timezone.utc) - timedelta(days=3)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 1
with self.assertRaises(StaleImageError) as ctx:
check_stale("agent image 'bot-bottle:latest'", created)
self.assertIn("agent image", str(ctx.exception))
self.assertIn("day(s) old", str(ctx.exception))
def test_naive_datetime_treated_as_utc(self):
# check_stale calls .astimezone(utc) on the input; naive datetimes
# that would be interpreted as local time should still work.
# We can't control the local tz in a unit test, so just ensure
# no exception is thrown for a very recent naive datetime.
naive_now = datetime(2099, 1, 1) # far future, always "fresh"
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 1
# Should not raise — the image is brand new.
check_stale("test image", naive_now)
class TestCheckStalePath(unittest.TestCase):
def test_delegates_to_check_stale_with_mtime(self):
with tempfile.NamedTemporaryFile() as f:
path = Path(f.name)
with patch("bot_bottle.image_cache.check_stale") as mock_check:
check_stale_path("some artifact", path)
mock_check.assert_called_once()
label, dt = mock_check.call_args.args
self.assertEqual("some artifact", label)
self.assertIsInstance(dt, datetime)
self.assertIsNotNone(dt.tzinfo)
def test_raises_stale_for_old_file(self):
with tempfile.NamedTemporaryFile() as f:
path = Path(f.name)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 0
with self.assertRaises(StaleImageError):
check_stale_path("cached artifact", path)
if __name__ == "__main__":
unittest.main()
+6 -85
View File
@@ -2,7 +2,6 @@
from __future__ import annotations
import dataclasses
import unittest
import tempfile
from pathlib import Path
@@ -16,7 +15,6 @@ from bot_bottle.backend.macos_container import launch
from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestIndex
_MANIFEST = ManifestIndex.from_json_obj({
@@ -83,7 +81,7 @@ def _plan(
provisioned_env={"CODEX_HOME": "/run/codex-home"},
)
return cast(MacosContainerBottlePlan, SimpleNamespace(
spec=SimpleNamespace(image_policy="fresh"),
spec=SimpleNamespace(),
manifest=_MANIFEST,
stage_dir=stage_dir,
slug="dev-abc",
@@ -294,7 +292,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
return MacosContainerBottlePlan(
spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
spec=cast(BottleSpec, SimpleNamespace()),
manifest=_MANIFEST,
stage_dir=stage_dir,
git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -339,9 +337,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
), patch.object(
launch.container_mod, "build_image", side_effect=fake_build,
), patch.object(launch, "info"):
images = launch.build_or_load_images(plan)
updated = launch._build_images(plan)
self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
self.assertEqual("bot-bottle-committed-dev-abc:latest", updated.image)
self.assertEqual(1, len(calls))
self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0])
@@ -360,89 +358,12 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
), patch.object(
launch.container_mod, "build_image", side_effect=fake_build,
):
images = launch.build_or_load_images(plan)
updated = launch._build_images(plan)
self.assertEqual("bot-bottle-agent:latest", images.agent)
self.assertEqual("bot-bottle-agent:latest", updated.image)
self.assertEqual(2, len(calls))
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_or_load_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_or_load_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"):
images = launch.build_or_load_images(plan)
build.assert_not_called()
self.assertEqual(plan.image, images.agent)
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"):
images = launch.build_or_load_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", images.agent)
if __name__ == "__main__":
unittest.main()
-66
View File
@@ -273,71 +273,5 @@ resolver #2
)
class TestMacosContainerImageCreatedAt(unittest.TestCase):
def _ok(self, stdout: str) -> "util.subprocess.CompletedProcess": # type: ignore
return util.subprocess.CompletedProcess(
args=[], returncode=0, stdout=stdout, stderr="",
)
def _fail(self, stderr: str = "no such image") -> "util.subprocess.CompletedProcess": # type: ignore
return util.subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr=stderr,
)
def test_parses_iso_timestamp_from_dict(self):
payload = '[{"created": "2025-06-01T12:00:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("bot-bottle-agent:latest")
self.assertEqual(2025, dt.year)
self.assertEqual(6, dt.month)
self.assertEqual(1, dt.day)
def test_accepts_list_or_dict_input(self):
# Container CLI may return a list; we take the first element.
payload = '[{"created": "2024-01-15T08:30:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("some-image:latest")
self.assertEqual(2024, dt.year)
def test_accepts_uppercase_Created_field(self):
payload = '[{"Created": "2024-03-20T10:00:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("some-image:latest")
self.assertEqual(2024, dt.year)
self.assertEqual(3, dt.month)
def test_dies_on_nonzero_returncode(self):
with patch.object(util.subprocess, "run", return_value=self._fail("not found")), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("missing:tag")
die.assert_called_once()
self.assertIn("missing:tag", die.call_args.args[0])
def test_dies_on_malformed_json(self):
with patch.object(util.subprocess, "run", return_value=self._ok("not-json {")), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
def test_dies_when_no_created_field(self):
payload = '[{"id": "sha256:abc123"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
self.assertIn("creation timestamp", die.call_args.args[0])
def test_dies_on_invalid_timestamp_format(self):
payload = '[{"created": "not-a-date"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
if __name__ == "__main__":
unittest.main()
@@ -187,54 +187,6 @@ class TestAgentFromPath(unittest.TestCase):
dockerfile="/repo/Dockerfile",
)
def test_cached_policy_uses_existing_artifact_without_build(self):
with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
digest = "abcdef0123456789"
artifact = Path(tmp) / f"{digest}.smolmachine.smolmachine"
artifact.write_text("")
plan = SimpleNamespace(
slug="dev-abc12",
agent_image="bot-bottle-claude:latest",
spec=SimpleNamespace(image_policy="cached"),
)
with patch.object(
_launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
), patch.object(
_launch_mod, "read_committed_image", return_value="",
), patch.object(
_launch_mod.docker_mod, "image_exists", return_value=True,
), patch.object(
_launch_mod.docker_mod, "image_id",
return_value=f"sha256:{digest}fffffffffffffffff",
), patch.object(
_launch_mod, "_ensure_smolmachine",
) as ensure:
result = _launch_mod._agent_from_path(cast(Any, plan))
self.assertEqual(artifact, result)
ensure.assert_not_called()
def test_cached_policy_dies_when_artifact_missing(self):
with tempfile.TemporaryDirectory(prefix="cached-smolmachine.") as tmp:
plan = SimpleNamespace(
slug="dev-abc12",
agent_image="bot-bottle-claude:latest",
spec=SimpleNamespace(image_policy="cached"),
)
with patch.object(
_launch_mod, "_SMOLMACHINE_CACHE_DIR", Path(tmp),
), patch.object(
_launch_mod, "read_committed_image", return_value="",
), patch.object(
_launch_mod.docker_mod, "image_exists", return_value=True,
), patch.object(
_launch_mod.docker_mod, "image_id",
return_value="sha256:abcdef0123456789fffffffffffffffff",
):
from bot_bottle.log import Die
with self.assertRaises(Die):
_launch_mod._agent_from_path(cast(Any, plan))
if __name__ == "__main__":
unittest.main()
@@ -313,9 +313,9 @@ class TestForceAllowlist(unittest.TestCase):
self.assertEqual(4, cfg["cpus"])
self.assertTrue(cfg["network"])
def test_patches_on_linux_too(self):
# force_allowlist no longer no-ops on Linux — the TSI
# allowlist must be enforced there as well.
def test_skips_patch_on_linux(self):
# On Linux, patching allowed_cidrs into the smolvm state DB
# crashes TSI boot. force_allowlist is a no-op on Linux.
with patch.object(loopback_alias, "_is_macos", return_value=False), \
patch.object(loopback_alias, "_SMOLVM_DB_PATH", self.db):
loopback_alias.force_allowlist("demo-vm", ["127.0.0.16/32"])
@@ -324,7 +324,7 @@ class TestForceAllowlist(unittest.TestCase):
"SELECT data FROM vms WHERE name='demo-vm'",
).fetchone()[0])
con.close()
self.assertEqual(["127.0.0.16/32"], cfg["allowed_cidrs"])
self.assertIsNone(cfg["allowed_cidrs"])
def test_skips_write_when_already_matching(self):
# A newer smolvm that honors --allow-cidr at create leaves the
+32 -6
View File
@@ -560,12 +560,13 @@ class TestLaunchResourceWiring(unittest.TestCase):
),
)
sidecar_artifact = Path("/cache/sidecar.smolmachine")
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(
@@ -575,10 +576,11 @@ class TestLaunchResourceWiring(unittest.TestCase):
"bot_bottle.backend.smolmachines.launch._forward.start_forwarder",
return_value=handle,
) as start_forwarder:
stamped = _launch._start_bundle(plan, "net", "127.0.0.16", sidecar_artifact, stack)
stamped = _launch._start_bundle(plan, "net", "127.0.0.16", stack)
ensure.assert_called_once()
start_vm.assert_called_once()
self.assertEqual(sidecar_artifact, start_vm.call_args.kwargs["from_path"])
self.assertEqual(Path("/cache/sidecar.smolmachine"), start_vm.call_args.kwargs["from_path"])
specs = start_forwarder.call_args.args[0]
self.assertEqual(
(
@@ -599,8 +601,11 @@ class TestLaunchResourceWiring(unittest.TestCase):
"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)
) as wait_exec_ready, patch(
"bot_bottle.backend.smolmachines.launch._loopback._is_macos",
return_value=True,
):
_launch._init_vm(plan, "127.0.0.16")
machine_exec.assert_called_once()
argv = machine_exec.call_args.args[1]
@@ -608,6 +613,27 @@ class TestLaunchResourceWiring(unittest.TestCase):
self.assertIn("chown -R node:node /home/node", argv[2])
wait_exec_ready.assert_called_once_with(plan.machine_name)
def test_init_vm_installs_iptables_on_linux(self):
plan = _plan()
from bot_bottle.backend.smolmachines.smolvm import SmolvmRunResult
with patch(
"bot_bottle.backend.smolmachines.launch._smolvm.machine_exec",
return_value=SmolvmRunResult(0, "", ""),
) as machine_exec, patch(
"bot_bottle.backend.smolmachines.launch._smolvm.wait_exec_ready",
), patch(
"bot_bottle.backend.smolmachines.launch._loopback._is_macos",
return_value=False,
):
_launch._init_vm(plan, "127.0.0.16")
# Two calls: ownership repair + iptables
self.assertEqual(2, machine_exec.call_count)
iptables_argv = machine_exec.call_args_list[1].args[1]
self.assertIn("iptables", iptables_argv[2])
self.assertIn("127.0.0.16/32", iptables_argv[2])
self.assertIn("127.0.0.0/8", iptables_argv[2])
class TestPortLabels(unittest.TestCase):
def test_known_and_dynamic_port_labels_round_trip(self):
-292
View File
@@ -1,292 +0,0 @@
"""Unit: stale-image check functions across backends, and the
BottleBackend.launch template method (prelaunch_checks + build_or_load_images).
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 — prelaunch_checks + _build_or_load_images
# ---------------------------------------------------------------------------
class TestBottleBackendLaunchTemplate(unittest.TestCase):
"""Verify the concrete launch() method on BottleBackend calls
_build_or_load_images and _launch_impl, and that prelaunch_checks is a no-op
on the base class."""
def _make_backend(self) -> Any:
from bot_bottle.backend.docker.backend import DockerBottleBackend
return DockerBottleBackend()
def test_launch_delegates_to_build_or_load_and_launch_impl(self) -> None:
from bot_bottle.backend import BottleImages
backend = self._make_backend()
plan = cast(Any, SimpleNamespace())
bottle = MagicMock()
images = BottleImages(agent="agent:latest", sidecar="sidecar:latest")
with patch.object(
backend, "_build_or_load_images", return_value=images,
) as build_mock, patch.object(
backend, "_launch_impl",
return_value=_bottle_cm(bottle),
) as impl_mock:
with backend.launch(plan):
pass
build_mock.assert_called_once_with(plan)
impl_mock.assert_called_once_with(plan, images)
def test_noop_default_prelaunch_checks(self) -> None:
from bot_bottle.backend.docker.backend import DockerBottleBackend
from bot_bottle.backend import BottleBackend
backend = DockerBottleBackend()
# Base-class prelaunch_checks is a no-op — must not raise.
BottleBackend.prelaunch_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_prelaunch_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.prelaunch_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_prelaunch_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.prelaunch_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_prelaunch_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.prelaunch_checks(plan) # type: ignore[arg-type]
sc.assert_called_once_with(plan)
if __name__ == "__main__":
unittest.main()