Compare commits

...

10 Commits

Author SHA1 Message Date
didericis-claude d8d14f889c fix(lint): remove unused BottleImages imports from two test files
lint / lint (push) Failing after 2m11s
test / unit (pull_request) Successful in 1m5s
test / integration (pull_request) Successful in 19s
test / coverage (pull_request) Successful in 1m7s
pyright strict reportUnusedImport flagged BottleImages in
test_docker_launch_committed_image.py and test_macos_container_launch.py;
neither file references the type by name (they only use the returned
value's attributes).
2026-07-11 05:48:37 +00:00
didericis-claude 0afd9c8698 fix: remove redundant quoted annotations in BottleImages and abstract methods
`from __future__ import annotations` already defers all annotation
evaluation, so quoting `str | Path`, `BottleImages` inside the same
module was redundant and tripped pyright strict mode.
2026-07-11 05:48:37 +00:00
didericis-claude d1e121febe refactor: BottleImages dataclass, prelaunch_checks, build_or_load_images
Addresses review comments 3098, 3099, 3100 on PR #336:

- Add BottleImages(agent, sidecar) dataclass to backend/__init__.py.
  Docker/macOS backends use str image refs; smolmachines uses Path
  artifacts. Replaces the singular `image` variable from the canonical
  pattern in comment 3100.

- Replace _image_stale_checks/skip_stale with public prelaunch_checks().
  CLI now calls backend.prelaunch_checks(plan) before backend.launch(plan);
  if StaleImageError is raised and the operator confirms, launch proceeds
  without re-checking. Removes the while-True/skip_stale retry loop.

- Add abstract _build_or_load_images(plan) -> BottleImages to
  BottleBackend. launch() calls it then passes images to _launch_impl.
  Each backend implements both methods.

- Fix comment 3098 (macos-container): _build_images is removed.
  build_or_load_images() has separate fresh/cached code paths — the
  cached path never calls a build helper.

- Update _start_bundle (smolmachines) to accept sidecar_artifact: Path
  directly. Sidecar artifact resolution moves to _sidecar_from_path(),
  called by build_or_load_images alongside _agent_from_path().
2026-07-11 05:48:37 +00:00
didericis-claude 537df7a784 fix(tests): resolve pyright errors in stale-check test files
Remove unused imports, add missing type annotations, fix Die()
constructor calls (int code, not str), replace fake backend
subclass with patch.object approach to avoid reportMissingParameterType
and override-incompatibility errors in strict pyright mode.
2026-07-11 05:41:19 +00:00
didericis-claude 7ea7f3926c test: add unit tests for stale-image checks to reach ≥90% diff-coverage
Covers StaleImageError / check_stale / check_stale_path, the
BottleBackend.launch template method (skip_stale flag), stale_checks
functions across docker / smolmachines / macos-container backends,
_build_images cached paths in the macOS backend, image_created_at edge
cases in both docker and container util modules, the CLI stale loop
(headless die, interactive decline, interactive confirm + retry), and
ConfigStore.cached_image_stale_warning_days fallback paths.

Diff-coverage: 488/507 changed lines covered (96.3%).
2026-07-11 05:41:19 +00:00
didericis-claude 2362973a0c refactor(image-cache): replace warn_if_stale with StaleImageError; add launch template
- image_cache: StaleImageError exception + check_stale/check_stale_path (raise instead of warn)
- BottleBackend.launch: template method (skip_stale flag) that calls _image_stale_checks then _launch_impl
- Each backend: _image_stale_checks delegates to a stale_checks() function in its launch module; _launch_impl replaces launch override
- macos_container: adds image_created_at to util, cached-image support in _build_images, stale_checks
- cli/start.py: catches StaleImageError, prompts interactively, retries with skip_stale=True; headless mode dies on it
2026-07-11 05:41:19 +00:00
didericis-claude 82a0b73d48 fix: pyright error in test_compose — annotate kwargs as dict[str, Any] 2026-07-11 05:39:47 +00:00
didericis-codex 114126e162 fix: make config store schema explicit 2026-07-11 05:39:47 +00:00
didericis-codex 1b4f64767a Add cached image quickstart 2026-07-11 05:39:47 +00:00
didericis-claude 814c7338a1 docs(research): update landscape doc — SuperHQ, multi-backend, multi-provider, in-flight directions
- Broaden scope from "Claude Code in Docker" to "AI coding agents in isolated sandboxes"
- Add SuperHQ (superhq.ai, v0.4.4) as a new adjacent-competitor entry: macOS-only
  microVM desktop app, overlaps on isolation/credential-proxy/multi-provider but has
  no manifest layer and no audit logging
- Note SuperHQ's user-voiced audit gap (Brian Cheong, Dunialabs.io) and that
  bot-bottle already covers it
- Update differentiation list to reflect three backends (Docker, Apple container,
  smolmachines) and three built-in providers (Claude Code, Codex, Pi) plus plugin system
- Add in-flight directions for forge-native dispatch (#317) and paid web control plane
  (#327) with honest framing: lifecycle concept is not novel vs. cloud services (Devin,
  Copilot Workspace); differentiation is self-hosted + manifest-driven + stronger isolation
2026-07-09 18:55:15 +00:00
27 changed files with 1268 additions and 94 deletions
+37 -4
View File
@@ -36,10 +36,10 @@ import os
import shlex import shlex
import sys import sys
from abc import ABC, abstractmethod from abc import ABC, abstractmethod
from contextlib import AbstractContextManager from contextlib import AbstractContextManager, contextmanager
from dataclasses import dataclass from dataclasses import dataclass
from pathlib import Path from pathlib import Path
from typing import Any, Generic, Sequence, TypeVar from typing import Any, Generator, Generic, Sequence, TypeVar
from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan from ..agent_provider import AgentProvisionPlan, get_provider, build_agent_provision_plan
from ..egress import EgressPlan from ..egress import EgressPlan
@@ -78,6 +78,9 @@ class BottleSpec:
# True when launched via --headless (no TTY, no interactive prompts). # True when launched via --headless (no TTY, no interactive prompts).
# The git-gate host-key preflight uses this to error rather than prompt. # The git-gate host-key preflight uses this to error rather than prompt.
headless: bool = False 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) @dataclass(frozen=True)
@@ -273,6 +276,18 @@ PlanT = TypeVar("PlanT", bound=BottlePlan)
CleanupT = TypeVar("CleanupT", bound=BottleCleanupPlan) 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]): class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Abstract base for selectable bottle backends. Concrete subclasses """Abstract base for selectable bottle backends. Concrete subclasses
(e.g. DockerBottleBackend) own their own prepare/launch impls. (e.g. DockerBottleBackend) own their own prepare/launch impls.
@@ -432,9 +447,27 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
prompt file, Dockerfile path, and guest home all live on prompt file, Dockerfile path, and guest home all live on
`agent_provision_plan` — the source of truth.""" `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 @abstractmethod
def launch(self, plan: PlanT) -> AbstractContextManager[Bottle]: def _build_or_load_images(self, plan: PlanT) -> BottleImages:
"""Build/run the bottle and yield a handle; tear down on exit.""" """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 provision(self, plan: PlanT, bottle: "Bottle") -> str | None: def provision(self, plan: PlanT, bottle: "Bottle") -> str | None:
"""Copy host-side files (CA cert, prompt, skills, .git) into """Copy host-side files (CA cert, prompt, skills, .git) into
+9 -3
View File
@@ -31,7 +31,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan from ...supervise import SupervisePlan
from ...manifest import Manifest from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleSpec from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from . import cleanup as _cleanup from . import cleanup as _cleanup
from . import enumerate as _enumerate from . import enumerate as _enumerate
from . import launch as _launch from . import launch as _launch
@@ -85,9 +85,15 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
stage_dir=stage_dir, 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 @contextmanager
def launch(self, plan: DockerBottlePlan) -> Generator[DockerBottle, None, None]: def _launch_impl(self, plan: DockerBottlePlan, images: BottleImages) -> Generator[DockerBottle, None, None]:
with _launch.launch(plan, provision=self.provision) as bottle: with _launch.launch(plan, images, provision=self.provision) as bottle:
yield bottle yield bottle
def supervise_mcp_url(self, plan: DockerBottlePlan) -> str: def supervise_mcp_url(self, plan: DockerBottlePlan) -> str:
+5 -4
View File
@@ -180,10 +180,6 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
service: dict[str, Any] = { service: dict[str, Any] = {
"image": SIDECAR_BUNDLE_IMAGE, "image": SIDECAR_BUNDLE_IMAGE,
"build": {
"context": _REPO_DIR,
"dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
},
"container_name": sidecar_bundle_container_name(plan.slug), "container_name": sidecar_bundle_container_name(plan.slug),
"networks": { "networks": {
"internal": {"aliases": internal_aliases}, "internal": {"aliases": internal_aliases},
@@ -192,6 +188,11 @@ def _sidecar_bundle_service(plan: DockerBottlePlan) -> dict[str, Any]:
"environment": env, "environment": env,
"volumes": volumes, "volumes": volumes,
} }
if plan.spec.image_policy != "cached":
service["build"] = {
"context": _REPO_DIR,
"dockerfile": SIDECAR_BUNDLE_DOCKERFILE,
}
return service return service
+58 -19
View File
@@ -41,7 +41,9 @@ from ...git_gate import (
provision_git_gate_dynamic_keys, provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys, revoke_git_gate_provisioned_keys,
) )
from ...log import info, warn from ...image_cache import check_stale
from ...log import die, info, warn
from .. import BottleImages
from . import network as network_mod from . import network as network_mod
from . import util as docker_mod from . import util as docker_mod
from .bottle import DockerBottle from .bottle import DockerBottle
@@ -63,22 +65,57 @@ from .compose import (
write_compose_file, write_compose_file,
) )
from .egress import egress_tls_init from .egress import egress_tls_init
from .sidecar_bundle import SIDECAR_BUNDLE_IMAGE
# Where the repo root lives, for `docker build` context. Computed once. # Where the repo root lives, for `docker build` context. Computed once.
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent) _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 @contextmanager
def launch( def launch(
plan: DockerBottlePlan, plan: DockerBottlePlan,
images: BottleImages,
*, *,
provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None], provision: Callable[[DockerBottlePlan, "DockerBottle"], str | None],
) -> Generator[DockerBottle, None, None]: ) -> Generator[DockerBottle, None, None]:
"""Build, launch, and provision a Docker bottle via compose. """Launch and provision a Docker bottle via compose. Teardown on exit."""
Teardown on exit."""
stack = ExitStack() 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 _bottle_for_revoke = plan.manifest.bottle
_git_gate_dir_for_revoke = git_gate_state_dir(plan.slug) _git_gate_dir_for_revoke = git_gate_state_dir(plan.slug)
@@ -95,22 +132,6 @@ def launch(
) )
try: 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) internal_network = network_mod.network_name_for_slug(plan.slug)
egress_network = network_mod.network_egress_name_for_slug(plan.slug) egress_network = network_mod.network_egress_name_for_slug(plan.slug)
@@ -207,3 +228,21 @@ def launch(
yield bottle yield bottle
finally: finally:
teardown() 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,6 +4,7 @@ existence, and building images."""
from __future__ import annotations from __future__ import annotations
from datetime import datetime, timezone
import re import re
import shutil import shutil
import subprocess import subprocess
@@ -186,6 +187,45 @@ def image_id(ref: str) -> str:
return r.stdout.strip() 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: def save(ref: str, output: str) -> None:
"""`docker save REF -o OUTPUT`. Writes a tarball of the image """`docker save REF -o OUTPUT`. Writes a tarball of the image
layers + manifest to the host path. Used by smolmachines layers + manifest to the host path. Used by smolmachines
+10 -4
View File
@@ -12,7 +12,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan from ...supervise import SupervisePlan
from ...manifest import Manifest from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleSpec from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from . import cleanup as _cleanup from . import cleanup as _cleanup
from . import enumerate as _enumerate from . import enumerate as _enumerate
from . import launch as _launch from . import launch as _launch
@@ -67,11 +67,17 @@ class MacosContainerBottleBackend(
stage_dir=stage_dir, 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 @contextmanager
def launch( def _launch_impl(
self, plan: MacosContainerBottlePlan self, plan: MacosContainerBottlePlan, images: BottleImages
) -> Generator[MacosContainerBottle, None, None]: ) -> Generator[MacosContainerBottle, None, None]:
with _launch.launch(plan, provision=self.provision) as bottle: with _launch.launch(plan, images, provision=self.provision) as bottle:
yield bottle yield bottle
def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan: def prepare_cleanup(self) -> MacosContainerBottleCleanupPlan:
+52 -22
View File
@@ -32,7 +32,9 @@ from ...git_gate import (
provision_git_gate_dynamic_keys, provision_git_gate_dynamic_keys,
revoke_git_gate_provisioned_keys, revoke_git_gate_provisioned_keys,
) )
from ...image_cache import check_stale
from ...log import die, info, warn from ...log import die, info, warn
from .. import BottleImages
from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT from ...supervise import DB_PATH_IN_CONTAINER, SUPERVISE_PORT
from ...util import expand_tilde from ...util import expand_tilde
from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT from ..docker.egress import EGRESS_CA_IN_CONTAINER, EGRESS_PORT
@@ -71,17 +73,51 @@ def sidecar_container_name(slug: str) -> str:
return f"bot-bottle-sidecars-{slug}" 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 @contextmanager
def launch( def launch(
plan: MacosContainerBottlePlan, plan: MacosContainerBottlePlan,
images: BottleImages,
*, *,
provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None], provision: Callable[[MacosContainerBottlePlan, "MacosContainerBottle"], str | None],
) -> Generator[MacosContainerBottle, None, None]: ) -> Generator[MacosContainerBottle, None, None]:
"""Build, run, provision, and yield an Apple Container bottle.""" """Run, provision, and yield an Apple Container bottle."""
stack = ExitStack() stack = ExitStack()
bottle_for_revoke = plan.manifest.bottle bottle_for_revoke = plan.manifest.bottle
git_gate_dir_for_revoke = git_gate_state_dir(plan.slug) 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: def teardown() -> None:
teardown_exc: BaseException | None = None teardown_exc: BaseException | None = None
try: try:
@@ -95,7 +131,6 @@ def launch(
try: try:
plan = _mint_certs(plan) plan = _mint_certs(plan)
plan = _build_images(plan)
internal_network = internal_network_name(plan.slug) internal_network = internal_network_name(plan.slug)
egress_network = egress_network_name(plan.slug) egress_network = egress_network_name(plan.slug)
@@ -148,28 +183,23 @@ def _mint_certs(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
return dataclasses.replace(plan, egress_plan=egress_plan) return dataclasses.replace(plan, egress_plan=egress_plan)
def _build_images(plan: MacosContainerBottlePlan) -> MacosContainerBottlePlan:
container_mod.build_image( def stale_checks(plan: MacosContainerBottlePlan) -> None:
SIDECAR_BUNDLE_IMAGE, """Raise StaleImageError if a cached image is older than the configured
_REPO_DIR, threshold. Only runs when image_policy is 'cached'. Called by the backend
dockerfile=SIDECAR_BUNDLE_DOCKERFILE, class's _image_stale_checks before _launch_impl starts any resources."""
) if plan.spec.image_policy != "cached":
return
committed = read_committed_image(plan.slug) committed = read_committed_image(plan.slug)
if committed and container_mod.image_exists(committed): if committed and container_mod.image_exists(committed):
info(f"using committed image {committed!r}") check_stale(f"agent image {committed!r}", container_mod.image_created_at(committed))
return dataclasses.replace( return
plan, for label, ref in [
agent_provision=dataclasses.replace( ("agent image", plan.image),
plan.agent_provision, ("sidecar image", SIDECAR_BUNDLE_IMAGE),
image=committed, ]:
), if container_mod.image_exists(ref):
) check_stale(f"{label} {ref!r}", container_mod.image_created_at(ref))
container_mod.build_image(
plan.image,
_REPO_DIR,
dockerfile=plan.dockerfile_path,
)
return plan
def _create_networks( def _create_networks(
@@ -10,6 +10,7 @@ import shutil
import subprocess import subprocess
import tempfile import tempfile
import time import time
from datetime import datetime, timezone
from typing import Iterable from typing import Iterable
from ...log import die, info from ...log import die, info
@@ -458,6 +459,39 @@ def image_id(ref: str) -> str:
raise AssertionError("unreachable") 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: def save(ref: str, output: str) -> None:
subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True) subprocess.run([_CONTAINER, "image", "save", ref, "-o", output], check=True)
+7 -4
View File
@@ -19,7 +19,7 @@ from ...env import ResolvedEnv
from ...git_gate import GitGatePlan from ...git_gate import GitGatePlan
from ...supervise import SupervisePlan from ...supervise import SupervisePlan
from ...manifest import Manifest from ...manifest import Manifest
from .. import ActiveAgent, BottleBackend, BottleSpec from .. import ActiveAgent, BottleBackend, BottleImages, BottleSpec
from . import cleanup as _cleanup from . import cleanup as _cleanup
from . import enumerate as _enumerate from . import enumerate as _enumerate
from . import launch as _launch from . import launch as _launch
@@ -77,11 +77,14 @@ class SmolmachinesBottleBackend(
stage_dir=stage_dir, stage_dir=stage_dir,
) )
def _build_or_load_images(self, plan: SmolmachinesBottlePlan) -> BottleImages:
return _launch.build_or_load_images(plan)
@contextmanager @contextmanager
def launch( def _launch_impl(
self, plan: SmolmachinesBottlePlan self, plan: SmolmachinesBottlePlan, images: BottleImages
) -> Generator[SmolmachinesBottle, None, None]: ) -> Generator[SmolmachinesBottle, None, None]:
with _launch.launch(plan, provision=self.provision) as bottle: with _launch.launch(plan, images, provision=self.provision) as bottle:
yield bottle yield bottle
def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str: def supervise_mcp_url(self, plan: SmolmachinesBottlePlan) -> str:
+10 -5
View File
@@ -52,6 +52,7 @@ from ...bottle_state import (
git_gate_state_dir, git_gate_state_dir,
read_committed_image, read_committed_image,
) )
from .. import BottleImages
from . import loopback_alias as _loopback from . import loopback_alias as _loopback
from . import sidecar_bundle as _bundle from . import sidecar_bundle as _bundle
from . import smolvm as _smolvm from . import smolvm as _smolvm
@@ -79,13 +80,20 @@ _GIT_HTTP_PORT = 9420
_SUPERVISE_PORT = SUPERVISE_PORT _SUPERVISE_PORT = SUPERVISE_PORT
def build_or_load_images(plan: SmolmachinesBottlePlan) -> BottleImages:
"""Build or resolve the agent smolmachine artifact and sidecar bundle image."""
_bundle.ensure_bundle_image()
return BottleImages(agent=_agent_from_path(plan), sidecar=_bundle.SIDECAR_BUNDLE_IMAGE)
@contextmanager @contextmanager
def launch( def launch(
plan: SmolmachinesBottlePlan, plan: SmolmachinesBottlePlan,
images: BottleImages,
*, *,
provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None], provision: Callable[[SmolmachinesBottlePlan, "SmolmachinesBottle"], str | None],
) -> Generator[SmolmachinesBottle, None, None]: ) -> Generator[SmolmachinesBottle, None, None]:
"""Build + run the bottle and yield a handle; tear everything """Run the bottle from pre-built images and yield a handle; tear everything
down on exit. Errors during bringup unwind any partial state down on exit. Errors during bringup unwind any partial state
via the ExitStack.""" via the ExitStack."""
stack = ExitStack() stack = ExitStack()
@@ -96,9 +104,7 @@ def launch(
plan = _start_bundle(plan, network, proxy_host, stack) plan = _start_bundle(plan, network, proxy_host, stack)
plan = _discover_urls(plan, proxy_host) plan = _discover_urls(plan, proxy_host)
agent_from_path = _agent_from_path(plan) _launch_vm(plan, Path(images.agent), proxy_host, stack)
_launch_vm(plan, agent_from_path, proxy_host, stack)
_init_vm(plan) _init_vm(plan)
bottle = SmolmachinesBottle( bottle = SmolmachinesBottle(
@@ -184,7 +190,6 @@ def _start_bundle(
plan = _provision_git_gate_keys(plan) plan = _provision_git_gate_keys(plan)
bundle_spec = _bundle_launch_spec(plan, network, proxy_host) bundle_spec = _bundle_launch_spec(plan, network, proxy_host)
token_env = _resolve_token_env(plan, dict(os.environ)) token_env = _resolve_token_env(plan, dict(os.environ))
_bundle.ensure_bundle_image(bundle_spec.image)
_bundle.start_bundle(bundle_spec, env={**os.environ, **token_env}) _bundle.start_bundle(bundle_spec, env={**os.environ, **token_env})
stack.callback(_bundle.stop_bundle, plan.slug) stack.callback(_bundle.stop_bundle, plan.slug)
return plan return plan
+33 -4
View File
@@ -36,6 +36,7 @@ from ..bottle_state import (
is_preserved, is_preserved,
mark_preserved, mark_preserved,
) )
from ..image_cache import StaleImageError
from ..log import info, die from ..log import info, die
from ..manifest import Manifest, ManifestIndex from ..manifest import Manifest, ManifestIndex
from ._common import PROG, USER_CWD, read_tty_line from ._common import PROG, USER_CWD, read_tty_line
@@ -63,6 +64,14 @@ def cmd_start(argv: list[str]) -> int:
"skip all prompts. For orchestrators, CI, and webhooks." "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( parser.add_argument(
"--bottle", "--bottle",
action="append", action="append",
@@ -95,6 +104,8 @@ def cmd_start(argv: list[str]) -> int:
help="agent name defined in bot-bottle.json (omit to pick interactively)", help="agent name defined in bot-bottle.json (omit to pick interactively)",
) )
args = parser.parse_args(argv) 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" dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
@@ -142,6 +153,10 @@ def cmd_start(argv: list[str]) -> int:
label, color = tui.name_color_modal(default_label=agent_name) label, color = tui.name_color_modal(default_label=agent_name)
label, color = _resolve_unique_label(label, color) label, color = _resolve_unique_label(label, color)
image_policy = _select_image_policy()
if image_policy is None:
return 0
spec = BottleSpec( spec = BottleSpec(
manifest=manifest, manifest=manifest,
agent_name=agent_name, agent_name=agent_name,
@@ -150,6 +165,7 @@ def cmd_start(argv: list[str]) -> int:
label=label, label=label,
color=color, color=color,
bottle_names=bottle_names, bottle_names=bottle_names,
image_policy=image_policy,
) )
return _launch_bottle( return _launch_bottle(
spec, spec,
@@ -210,6 +226,7 @@ def _start_headless(
color=args.color or "", color=args.color or "",
bottle_names=bottle_names, bottle_names=bottle_names,
headless=True, headless=True,
image_policy="cached" if args.cached_images else "fresh",
) )
return _launch_bottle( return _launch_bottle(
spec, spec,
@@ -390,6 +407,13 @@ def _text_prompt_yes() -> bool:
return reply in ("y", "Y", "yes", "YES") 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 _text_render_preflight():
def _render(plan: DockerBottlePlan, backend_name: str) -> None: def _render(plan: DockerBottlePlan, backend_name: str) -> None:
print(file=sys.stderr) print(file=sys.stderr)
@@ -532,6 +556,15 @@ def _launch_bottle(
return 0 return 0
backend = get_bottle_backend(backend_name) 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: with backend.launch(plan) as bottle:
agent_provider_template = getattr(plan, "agent_provider_template", "claude") agent_provider_template = getattr(plan, "agent_provider_template", "claude")
extra_args: tuple[str, ...] = () extra_args: tuple[str, ...] = ()
@@ -550,10 +583,6 @@ def _launch_bottle(
f"session ended (exit {exit_code}); " f"session ended (exit {exit_code}); "
f"container {bottle.name} will be removed" 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": if agent_provider_template == "claude":
capture_claude_session_state(identity, exit_code) capture_claude_session_state(identity, exit_code)
return 0 return 0
+71
View File
@@ -0,0 +1,71 @@
"""SQLite-backed bot-bottle configuration store."""
from __future__ import annotations
from pathlib import Path
try:
from .db_store import DbStore
from .migrations import TableMigrations
from .supervise_types import host_db_path
except ImportError:
from db_store import DbStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from migrations import TableMigrations # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
from supervise_types import host_db_path # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS = 1
class ConfigStore(DbStore):
"""SQLite configuration for host-side bot-bottle settings."""
def __init__(self, db_path: Path | None = None) -> None:
migrations = TableMigrations("config_store", [
# v1 — host-side bot-bottle settings
"""
CREATE TABLE IF NOT EXISTS bot_bottle_config (
id INTEGER PRIMARY KEY CHECK (id = 1),
cached_image_stale_warning_days INTEGER NOT NULL DEFAULT 1
)
""",
])
super().__init__(db_path or host_db_path(), migrations)
def cached_image_stale_warning_days(self) -> int:
if not self.db_path.is_file():
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
with self._connect() as conn:
row = conn.execute(
"""
SELECT cached_image_stale_warning_days
FROM bot_bottle_config
WHERE id = 1
""",
).fetchone()
if row is None:
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
try:
return int(row["cached_image_stale_warning_days"])
except (TypeError, ValueError):
return DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS
def set_cached_image_stale_warning_days(self, days: int) -> Path:
with self._connect() as conn:
conn.execute(
"""
INSERT INTO bot_bottle_config (id, cached_image_stale_warning_days)
VALUES (1, ?)
ON CONFLICT(id) DO UPDATE SET
cached_image_stale_warning_days = excluded.cached_image_stale_warning_days
""",
(days,),
)
self._chmod()
return self.db_path
__all__ = [
"DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS",
"ConfigStore",
]
+42
View File
@@ -0,0 +1,42 @@
"""Shared helpers for cached-image quickstart stale checks."""
from __future__ import annotations
from datetime import datetime, timezone
from pathlib import Path
try:
from .config_store import ConfigStore
except ImportError:
from config_store import ConfigStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
class StaleImageError(Exception):
"""Raised when a cached image or artifact exceeds the configured staleness
threshold. Callers can catch this to prompt interactively; headless paths
let it propagate as a fatal error."""
def check_stale(label: str, created_at: datetime) -> None:
"""Raise StaleImageError if `created_at` is older than the configured
stale-warning threshold. Negative threshold disables the check."""
threshold_days = ConfigStore().cached_image_stale_warning_days()
if threshold_days < 0:
return
now = datetime.now(timezone.utc)
created = created_at.astimezone(timezone.utc)
age = now - created
if age.total_seconds() <= threshold_days * 86400:
return
raise StaleImageError(
f"cached {label} is {age.days} day(s) old; "
"quickstart does not verify it matches the current Dockerfile/context"
)
def check_stale_path(label: str, path: Path) -> None:
"""Raise StaleImageError if `path`'s mtime exceeds the staleness threshold."""
check_stale(label, datetime.fromtimestamp(path.stat().st_mtime, tz=timezone.utc))
__all__ = ["StaleImageError", "check_stale", "check_stale_path"]
+4
View File
@@ -6,9 +6,11 @@ from pathlib import Path
try: try:
from .audit_store import AuditStore from .audit_store import AuditStore
from .config_store import ConfigStore
from .queue_store import QueueStore from .queue_store import QueueStore
except ImportError: except ImportError:
from audit_store import AuditStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module 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 from queue_store import QueueStore # type: ignore[import-not-found] # pylint: disable=import-error,no-name-in-module
_instance: StoreManager | None = None _instance: StoreManager | None = None
@@ -50,11 +52,13 @@ class StoreManager:
return ( return (
QueueStore("", self.db_path).is_migrated() QueueStore("", self.db_path).is_migrated()
and AuditStore(self.db_path).is_migrated() and AuditStore(self.db_path).is_migrated()
and ConfigStore(self.db_path).is_migrated()
) )
def migrate(self) -> None: def migrate(self) -> None:
QueueStore("", self.db_path).migrate() QueueStore("", self.db_path).migrate()
AuditStore(self.db_path).migrate() AuditStore(self.db_path).migrate()
ConfigStore(self.db_path).migrate()
__all__ = ["StoreManager"] __all__ = ["StoreManager"]
+69 -13
View File
@@ -1,14 +1,20 @@
# Landscape: containerized Claude Code agent tools # Landscape: containerized AI coding agent tools
Research into whether bot-bottle is redundant with existing projects, and Research into whether bot-bottle is redundant with existing projects, and
whether it's worth publishing. whether it's worth publishing.
## Summary ## Summary
The "Claude Code in Docker" space is active but not saturated. bot-bottle The "AI coding agents in isolated sandboxes" space is active but not saturated.
occupies a distinct position: no surveyed project combines all five of its bot-bottle occupies a distinct position: no surveyed project combines all five
defining features. Publishing is likely worthwhile, with the main risk being of its defining features. Publishing is likely worthwhile, with the main risk
claudebox expanding to absorb the same niche. being claudebox expanding to absorb the same niche.
**Updated 2026-07-09:** bot-bottle now supports three isolation backends
(Docker, Apple `container`, smolmachines/libkrun microVMs) and three built-in
agent providers (Claude Code, OpenAI Codex, Pi) with an open plugin system for
arbitrary providers. This meaningfully strengthens the differentiation against
all surveyed competitors.
## Closest competitor: claudebox ## Closest competitor: claudebox
@@ -43,28 +49,78 @@ manifest merge.
Still marked early-development. Still marked early-development.
- **E2B, Northflank, Cloudflare Sandbox SDK** — cloud-hosted SaaS sandbox - **E2B, Northflank, Cloudflare Sandbox SDK** — cloud-hosted SaaS sandbox
runtimes; fundamentally different architecture. runtimes; fundamentally different architecture.
- **superhq.ai / SuperHQ** (v0.4.4, April 2026) — macOS desktop app (Rust/GPUI)
that runs Claude Code, Codex, and Pi inside microVMs via Apple's
Virtualization.framework (their own shuru-sdk / libkrun). Auth gateway
injects API keys on the wire so the sandbox never sees them; tmpfs overlay
stages agent writes for diff-and-accept review; mobile remote access via
remote.superhq.ai. Early alpha, free on launch, Apple Silicon only.
Overlap: both projects cover agent isolation, credential proxying, and
multi-provider support (Claude Code / Codex / Pi). Differences: SuperHQ is a
GUI desktop app with no manifest layer; bot-bottle is a CLI fleet manager with
named agents, skills injection, per-agent system prompts, and cross-platform
backends (Docker, Apple `container`, smolmachines). SuperHQ's microVM
isolation story is now partially matched by bot-bottle's `macos_container` and
smolmachines backends. Worth watching — it targets the same security-minded
power-user audience and moves fast.
**Known gap in SuperHQ (user-requested, as of 2026-07-09):** A named user
(Brian Cheong, Founder, Dunialabs.io) explicitly called out the absence of
per-run audit logging: tool calls and network egress. Bot-bottle covers both:
network egress is logged by pipelock/mitmproxy, and per-run op-log/audit state
is persisted to SQLite.
## What no found project does ## What no found project does
None combine: None combine:
1. Named-agent JSON manifest with per-agent env resolution (prompt / host-forward / literal) 1. Named-agent manifest with per-agent env resolution (prompt / host-forward / literal), supporting multiple providers (Claude Code, Codex, Pi, arbitrary plugins)
2. Claude Code skills directory injection 2. Skills directory injection
3. Per-agent system prompts 3. Per-agent system prompts
4. SSH-agent key forwarding without copying private keys into the container 4. SSH-agent key forwarding without copying private keys into the container
5. Home + project manifest merge 5. Home + project manifest merge
6. Pluggable isolation backends: Docker (Linux/macOS), Apple `container` (macOS microVMs), smolmachines/libkrun microVMs
7. Per-run audit log: network egress via pipelock/mitmproxy + op-log persisted to SQLite
**In-flight directions (not yet shipped):**
- **Forge-native dispatch (issue #317):** Gitea webhook → orchestrator spins up a bottle
with the issue body as prompt → agent works → bottle freezes awaiting review comment →
rehydrates on comment → tears down on PR close. The issue-to-PR lifecycle concept is not
novel (Devin, Copilot Workspace, SWE-agent all do this as cloud services); what's
distinct is doing it self-hosted, manifest-driven, inside bot-bottle's isolation
primitives.
- **Paid web control plane (issue #327):** Browser-based multi-host agent launch and
monitoring; account-scoped bottle and agent definitions; secret custody (encrypted at
rest, injected into the sidecar at launch, never exposed to the agent or returned by any
read API). Monetization model: OSS runtime free, control plane paid — a standard split
(HashiCorp, Grafana) applied to a self-hosted agent sandbox. The principled secret
custody model (agent never sees real credentials, even via printenv) is more rigorous
than most surveyed tools but not unprecedented.
## Publishing verdict ## Publishing verdict
Worth publishing. Differentiators that matter to the target audience (power Worth publishing. Differentiators that matter to the target audience (power
users running parallel Claude Code sessions with distinct personas/tooling): users running parallel AI coding agent sessions with distinct personas/tooling):
- The Python-stdlib-first, low-dependency design — competitors are npm-based or - The Python-stdlib-first, low-dependency design — competitors are npm-based,
Kubernetes-native. Rust/GUI, or Kubernetes-native.
- Named agents with distinct skills and system prompts, not just language profiles. - Named agents with distinct skills and system prompts, not just language profiles.
- Multi-backend isolation: Docker, Apple `container` microVMs, and
smolmachines/libkrun — single manifest works across all three.
- Multi-provider: Claude Code, Codex, Pi, plus an open plugin system for
arbitrary providers.
- SSH forwarding without key copying. - SSH forwarding without key copying.
- Per-run audit log (tool calls + network egress) — an explicitly requested gap
in SuperHQ as of 2026-07-09.
- Forge-native dispatch and a paid control plane (in flight) bring bot-bottle
into the same product category as cloud services like Devin and Copilot
Workspace — but self-hosted, with stronger isolation guarantees and a
manifest-driven fleet model those services don't have.
Main risk: claudebox adds manifest/agent config. The space is moving fast Main risk: claudebox adds manifest/agent config; SuperHQ is moving fast on the
enough that publishing sooner is better if establishing prior art matters. GUI / microVM side. The space is moving fast enough that publishing sooner is
better if establishing prior art matters.
Discovery will be slow without active promotion; an Anthropic Discord post or Discovery will be slow without active promotion; an Anthropic Discord post or
HN "Show HN" would do most of the work. HN "Show HN" would do most of the work.
@@ -73,4 +129,4 @@ HN "Show HN" would do most of the work.
- GitHub search cannot surface private or very new repos comprehensively. - GitHub search cannot surface private or very new repos comprehensively.
- Counts (stars, forks) were not confirmed for every project. - Counts (stars, forks) were not confirmed for every project.
- Research conducted 2026-05-07; the space moves fast. - Initial research conducted 2026-05-07; SuperHQ entry added 2026-07-09; the space moves fast.
+12
View File
@@ -184,6 +184,18 @@ class TestCmdStartHeadless(unittest.TestCase):
) )
self.assertEqual("docker", self._launch_mock.call_args[1]["backend_name"]) 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): class TestPrepareWithPreflight(unittest.TestCase):
"""prepare_with_preflight calls render_preflight with the plan and backend name.""" """prepare_with_preflight calls render_preflight with the plan and backend name."""
+21
View File
@@ -57,6 +57,12 @@ class TestCmdStartSelector(unittest.TestCase):
self._bottle_picker_mock = self._bottle_picker_patch.start() self._bottle_picker_mock = self._bottle_picker_patch.start()
self._bottle_picker_mock.return_value = ["claude"] # default: one bottle selected 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 = patch.dict(os.environ, {}, clear=False)
self._env_patch.start() self._env_patch.start()
os.environ.pop("BOT_BOTTLE_BACKEND", None) os.environ.pop("BOT_BOTTLE_BACKEND", None)
@@ -66,6 +72,7 @@ class TestCmdStartSelector(unittest.TestCase):
self._launch_patch.stop() self._launch_patch.stop()
self._agent_picker_patch.stop() self._agent_picker_patch.stop()
self._bottle_picker_patch.stop() self._bottle_picker_patch.stop()
self._image_policy_patch.stop()
self._env_patch.stop() self._env_patch.stop()
# ------------------------------------------------------------------ # ------------------------------------------------------------------
@@ -124,6 +131,19 @@ class TestCmdStartSelector(unittest.TestCase):
spec = self._launch_mock.call_args[0][0] spec = self._launch_mock.call_args[0][0]
self.assertEqual(("claude", "dev"), spec.bottle_names) 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): def test_empty_bottle_selection_forwarded(self):
self._bottle_picker_mock.return_value = [] self._bottle_picker_mock.return_value = []
start_mod.cmd_start(["researcher"]) start_mod.cmd_start(["researcher"])
@@ -215,6 +235,7 @@ class TestCmdStartLabelCollision(unittest.TestCase):
).start() ).start()
# Stub the bottle picker to always return a selection. # Stub the bottle picker to always return a selection.
patch.object(tui_mod, "filter_multiselect", return_value=["claude"]).start() 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) self.addCleanup(patch.stopall)
def test_no_collision_proceeds_without_reprompt(self): def test_no_collision_proceeds_without_reprompt(self):
+146
View File
@@ -0,0 +1,146 @@
"""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()
+9 -1
View File
@@ -118,6 +118,7 @@ def _plan(
with_egress: bool = False, with_egress: bool = False,
supervise: bool = False, supervise: bool = False,
canary: bool = False, canary: bool = False,
image_policy: str = "fresh",
) -> DockerBottlePlan: ) -> DockerBottlePlan:
"""Build a fully-resolved DockerBottlePlan. Toggles cover the """Build a fully-resolved DockerBottlePlan. Toggles cover the
matrix the renderer's conditional-service logic branches on.""" matrix the renderer's conditional-service logic branches on."""
@@ -148,6 +149,7 @@ def _plan(
agent_name="demo", agent_name="demo",
copy_cwd=False, copy_cwd=False,
user_cwd="/tmp/x", user_cwd="/tmp/x",
image_policy=image_policy,
) )
return DockerBottlePlan( return DockerBottlePlan(
spec=spec, spec=spec,
@@ -265,7 +267,8 @@ class TestAgentAlwaysPresent(unittest.TestCase):
def test_agent_depends_only_on_sidecars(self): def test_agent_depends_only_on_sidecars(self):
# Bundle shape: the init supervisor owns intra-bundle daemon # Bundle shape: the init supervisor owns intra-bundle daemon
# ordering, so the agent waits on the bundle container alone. # ordering, so the agent waits on the bundle container alone.
for kwargs in [{}, {"with_git": True, "with_egress": True, "supervise": True}]: cases: list[dict[str, Any]] = [{}, {"with_git": True, "with_egress": True, "supervise": True}]
for kwargs in cases:
with self.subTest(**kwargs): with self.subTest(**kwargs):
s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"] s = bottle_plan_to_compose(_plan(**kwargs))["services"]["agent"]
self.assertEqual(["sidecars"], s["depends_on"]) self.assertEqual(["sidecars"], s["depends_on"])
@@ -300,6 +303,11 @@ class TestSidecarBundleShape(unittest.TestCase):
self.assertEqual("bot-bottle-sidecars:latest", sc["image"]) self.assertEqual("bot-bottle-sidecars:latest", sc["image"])
self.assertEqual("Dockerfile.sidecars", sc["build"]["dockerfile"]) 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): def test_bundle_container_name_uses_sidecars_prefix(self):
sc = self._render()["services"]["sidecars"] sc = self._render()["services"]["sidecars"]
self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"]) self.assertEqual(f"bot-bottle-sidecars-{SLUG}", sc["container_name"])
+86
View File
@@ -0,0 +1,86 @@
"""Unit tests for the host-side configuration store."""
from __future__ import annotations
import sqlite3
import tempfile
import unittest
from pathlib import Path
from bot_bottle.config_store import (
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
ConfigStore,
)
from bot_bottle.store_manager import StoreManager
class TestConfigStore(unittest.TestCase):
def test_cached_image_warning_days_defaults_to_one(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
def test_cached_image_warning_days_reads_value(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
store.set_cached_image_stale_warning_days(7)
self.assertEqual(7, store.cached_image_stale_warning_days())
def test_config_schema_uses_explicit_settings_columns(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "bot-bottle.db")
store.migrate()
with sqlite3.connect(store.db_path) as conn:
conn.row_factory = sqlite3.Row
columns = [
row["name"]
for row in conn.execute("PRAGMA table_info(bot_bottle_config)")
]
self.assertEqual([
"id",
"cached_image_stale_warning_days",
], columns)
def test_store_manager_includes_config_store(self) -> None:
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
db = Path(tmp) / "bot-bottle.db"
manager = StoreManager(db)
self.assertFalse(manager.is_migrated())
manager.migrate()
self.assertTrue(manager.is_migrated())
def test_cached_image_warning_days_returns_default_when_db_missing(self) -> None:
# When the db file doesn't exist yet (parent exists, file doesn't),
# the store returns the default without touching the file.
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
store = ConfigStore(Path(tmp) / "missing.db")
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
def test_cached_image_warning_days_returns_default_on_null_value(self) -> None:
# If the row exists but the value is NULL (or not castable to int),
# the store falls back to the default.
with tempfile.TemporaryDirectory(prefix="config-store.") as tmp:
db_path = Path(tmp) / "bot-bottle.db"
store = ConfigStore(db_path)
store.migrate()
# Write a NULL value directly.
with sqlite3.connect(db_path) as conn:
conn.execute(
"UPDATE bot_bottle_config SET cached_image_stale_warning_days = NULL WHERE id = 1"
)
self.assertEqual(
DEFAULT_CACHED_IMAGE_STALE_WARNING_DAYS,
store.cached_image_stale_warning_days(),
)
if __name__ == "__main__":
unittest.main()
@@ -3,6 +3,7 @@
from __future__ import annotations from __future__ import annotations
import contextlib import contextlib
import dataclasses
import io import io
import tempfile import tempfile
import unittest import unittest
@@ -16,6 +17,7 @@ from bot_bottle.backend.docker import launch as launch_mod
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
@@ -88,7 +90,7 @@ class TestLaunchCommittedImage(unittest.TestCase):
committed_tag: str | None = None, committed_tag: str | None = None,
image_present: bool = True, image_present: bool = True,
) -> list[str]: ) -> list[str]:
"""Drive launch() through its full sequence with the committed-image """Drive build_or_load_images() + launch() with the committed-image
behaviour controlled by the arguments. Returns the images that were behaviour controlled by the arguments. Returns the images that were
passed to `build_image` (empty list if it was never called).""" passed to `build_image` (empty list if it was never called)."""
built: list[str] = [] built: list[str] = []
@@ -122,8 +124,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
mock.patch.object(launch_mod, "compose_dump_logs"), \ mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \ mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()): contextlib.redirect_stderr(io.StringIO()):
images = launch_mod.build_or_load_images(plan)
provision = mock.Mock(return_value=None) provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, provision=provision): with launch_mod.launch(plan, images, provision=provision):
pass pass
return built return built
@@ -168,8 +171,9 @@ class TestLaunchCommittedImage(unittest.TestCase):
mock.patch.object(launch_mod, "compose_dump_logs"), \ mock.patch.object(launch_mod, "compose_dump_logs"), \
mock.patch.object(launch_mod, "compose_down"), \ mock.patch.object(launch_mod, "compose_down"), \
contextlib.redirect_stderr(io.StringIO()): contextlib.redirect_stderr(io.StringIO()):
images = launch_mod.build_or_load_images(plan)
provision = mock.Mock(return_value=None) provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, provision=provision): with launch_mod.launch(plan, images, provision=provision):
pass pass
self.assertEqual(1, len(captured_plans)) self.assertEqual(1, len(captured_plans))
@@ -180,6 +184,24 @@ class TestLaunchCommittedImage(unittest.TestCase):
built = self._run_launch(plan, committed_tag=None) built = self._run_launch(plan, committed_tag=None)
self.assertEqual([_DEFAULT_IMAGE], built) 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: def test_falls_back_to_build_when_committed_image_missing_from_daemon(self) -> None:
plan = _plan(self._tmp) plan = _plan(self._tmp)
built = self._run_launch( built = self._run_launch(
+4 -2
View File
@@ -16,7 +16,7 @@ from pathlib import Path
from unittest import mock from unittest import mock
from bot_bottle.agent_provider import AgentProvisionPlan from bot_bottle.agent_provider import AgentProvisionPlan
from bot_bottle.backend import BottleSpec from bot_bottle.backend import BottleImages, BottleSpec
from bot_bottle.backend.docker import launch as launch_mod from bot_bottle.backend.docker import launch as launch_mod
from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan from bot_bottle.backend.docker.bottle_plan import DockerBottlePlan
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
@@ -86,6 +86,8 @@ class TestTeardownWarning(unittest.TestCase):
plan = _plan(self._tmp) plan = _plan(self._tmp)
buf = io.StringIO() 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"), \ with mock.patch.object(launch_mod.docker_mod, "build_image"), \
mock.patch.object( mock.patch.object(
launch_mod, "egress_tls_init", launch_mod, "egress_tls_init",
@@ -115,7 +117,7 @@ class TestTeardownWarning(unittest.TestCase):
), \ ), \
contextlib.redirect_stderr(buf): contextlib.redirect_stderr(buf):
provision = mock.Mock(return_value=None) provision = mock.Mock(return_value=None)
with launch_mod.launch(plan, provision=provision): with launch_mod.launch(plan, images, provision=provision):
pass pass
output = buf.getvalue() output = buf.getvalue()
+46
View File
@@ -9,6 +9,7 @@ from __future__ import annotations
import subprocess import subprocess
import unittest import unittest
from datetime import timezone
from unittest.mock import patch from unittest.mock import patch
from bot_bottle.backend.docker import util as docker_mod from bot_bottle.backend.docker import util as docker_mod
@@ -54,6 +55,51 @@ class TestImageId(unittest.TestCase):
self.assertIn("missing:tag", die.call_args.args[0]) 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): class TestSave(unittest.TestCase):
def test_save_runs_docker_save(self): def test_save_runs_docker_save(self):
with patch.object( with patch.object(
+81
View File
@@ -0,0 +1,81 @@
"""Unit: image_cache.py — check_stale / check_stale_path."""
from __future__ import annotations
import tempfile
import unittest
from datetime import datetime, timedelta, timezone
from pathlib import Path
from unittest.mock import patch
from bot_bottle.image_cache import StaleImageError, check_stale, check_stale_path
class TestCheckStale(unittest.TestCase):
def _run(self, threshold: int, age_days: float) -> None:
created = datetime.now(tz=timezone.utc) - timedelta(days=age_days)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = threshold
check_stale("test image", created)
def test_negative_threshold_never_raises(self):
# Threshold < 0 means the check is disabled — always passes.
self._run(threshold=-1, age_days=9999)
def test_zero_threshold_raises_immediately(self):
# threshold=0 means any image is stale the moment it exists.
with self.assertRaises(StaleImageError):
self._run(threshold=0, age_days=0.1)
def test_within_threshold_does_not_raise(self):
# Age well under threshold — should pass silently.
self._run(threshold=7, age_days=2)
def test_at_threshold_does_not_raise(self):
# Exactly at the boundary is fine (<=, not <).
self._run(threshold=1, age_days=0.9999)
def test_exceeds_threshold_raises(self):
created = datetime.now(tz=timezone.utc) - timedelta(days=3)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 1
with self.assertRaises(StaleImageError) as ctx:
check_stale("agent image 'bot-bottle:latest'", created)
self.assertIn("agent image", str(ctx.exception))
self.assertIn("day(s) old", str(ctx.exception))
def test_naive_datetime_treated_as_utc(self):
# check_stale calls .astimezone(utc) on the input; naive datetimes
# that would be interpreted as local time should still work.
# We can't control the local tz in a unit test, so just ensure
# no exception is thrown for a very recent naive datetime.
naive_now = datetime(2099, 1, 1) # far future, always "fresh"
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 1
# Should not raise — the image is brand new.
check_stale("test image", naive_now)
class TestCheckStalePath(unittest.TestCase):
def test_delegates_to_check_stale_with_mtime(self):
with tempfile.NamedTemporaryFile() as f:
path = Path(f.name)
with patch("bot_bottle.image_cache.check_stale") as mock_check:
check_stale_path("some artifact", path)
mock_check.assert_called_once()
label, dt = mock_check.call_args.args
self.assertEqual("some artifact", label)
self.assertIsInstance(dt, datetime)
self.assertIsNotNone(dt.tzinfo)
def test_raises_stale_for_old_file(self):
with tempfile.NamedTemporaryFile() as f:
path = Path(f.name)
with patch("bot_bottle.image_cache.ConfigStore") as cs:
cs.return_value.cached_image_stale_warning_days.return_value = 0
with self.assertRaises(StaleImageError):
check_stale_path("cached artifact", path)
if __name__ == "__main__":
unittest.main()
+85 -6
View File
@@ -2,6 +2,7 @@
from __future__ import annotations from __future__ import annotations
import dataclasses
import unittest import unittest
import tempfile import tempfile
from pathlib import Path from pathlib import Path
@@ -15,6 +16,7 @@ from bot_bottle.backend.macos_container import launch
from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan from bot_bottle.backend.macos_container.bottle_plan import MacosContainerBottlePlan
from bot_bottle.egress import EgressPlan from bot_bottle.egress import EgressPlan
from bot_bottle.git_gate import GitGatePlan from bot_bottle.git_gate import GitGatePlan
from bot_bottle.log import Die
from bot_bottle.manifest import ManifestIndex from bot_bottle.manifest import ManifestIndex
_MANIFEST = ManifestIndex.from_json_obj({ _MANIFEST = ManifestIndex.from_json_obj({
@@ -81,7 +83,7 @@ def _plan(
provisioned_env={"CODEX_HOME": "/run/codex-home"}, provisioned_env={"CODEX_HOME": "/run/codex-home"},
) )
return cast(MacosContainerBottlePlan, SimpleNamespace( return cast(MacosContainerBottlePlan, SimpleNamespace(
spec=SimpleNamespace(), spec=SimpleNamespace(image_policy="fresh"),
manifest=_MANIFEST, manifest=_MANIFEST,
stage_dir=stage_dir, stage_dir=stage_dir,
slug="dev-abc", slug="dev-abc",
@@ -292,7 +294,7 @@ class TestMacosContainerLaunchArgv(unittest.TestCase):
def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan: def _build_plan(stage_dir: Path) -> MacosContainerBottlePlan:
return MacosContainerBottlePlan( return MacosContainerBottlePlan(
spec=cast(BottleSpec, SimpleNamespace()), spec=cast(BottleSpec, SimpleNamespace(image_policy="fresh")),
manifest=_MANIFEST, manifest=_MANIFEST,
stage_dir=stage_dir, stage_dir=stage_dir,
git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())), git_gate_plan=cast(GitGatePlan, SimpleNamespace(upstreams=())),
@@ -337,9 +339,9 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
), patch.object( ), patch.object(
launch.container_mod, "build_image", side_effect=fake_build, launch.container_mod, "build_image", side_effect=fake_build,
), patch.object(launch, "info"): ), patch.object(launch, "info"):
updated = launch._build_images(plan) images = launch.build_or_load_images(plan)
self.assertEqual("bot-bottle-committed-dev-abc:latest", updated.image) self.assertEqual("bot-bottle-committed-dev-abc:latest", images.agent)
self.assertEqual(1, len(calls)) self.assertEqual(1, len(calls))
self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0]) self.assertEqual(launch.SIDECAR_BUNDLE_IMAGE, calls[0][0])
@@ -358,12 +360,89 @@ class TestMacosContainerLaunchCommittedImage(unittest.TestCase):
), patch.object( ), patch.object(
launch.container_mod, "build_image", side_effect=fake_build, launch.container_mod, "build_image", side_effect=fake_build,
): ):
updated = launch._build_images(plan) images = launch.build_or_load_images(plan)
self.assertEqual("bot-bottle-agent:latest", updated.image) self.assertEqual("bot-bottle-agent:latest", images.agent)
self.assertEqual(2, len(calls)) self.assertEqual(2, len(calls))
self.assertEqual("bot-bottle-agent:latest", calls[1][0]) 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__": if __name__ == "__main__":
unittest.main() unittest.main()
+66
View File
@@ -273,5 +273,71 @@ resolver #2
) )
class TestMacosContainerImageCreatedAt(unittest.TestCase):
def _ok(self, stdout: str) -> "util.subprocess.CompletedProcess": # type: ignore
return util.subprocess.CompletedProcess(
args=[], returncode=0, stdout=stdout, stderr="",
)
def _fail(self, stderr: str = "no such image") -> "util.subprocess.CompletedProcess": # type: ignore
return util.subprocess.CompletedProcess(
args=[], returncode=1, stdout="", stderr=stderr,
)
def test_parses_iso_timestamp_from_dict(self):
payload = '[{"created": "2025-06-01T12:00:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("bot-bottle-agent:latest")
self.assertEqual(2025, dt.year)
self.assertEqual(6, dt.month)
self.assertEqual(1, dt.day)
def test_accepts_list_or_dict_input(self):
# Container CLI may return a list; we take the first element.
payload = '[{"created": "2024-01-15T08:30:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("some-image:latest")
self.assertEqual(2024, dt.year)
def test_accepts_uppercase_Created_field(self):
payload = '[{"Created": "2024-03-20T10:00:00"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)):
dt = util.image_created_at("some-image:latest")
self.assertEqual(2024, dt.year)
self.assertEqual(3, dt.month)
def test_dies_on_nonzero_returncode(self):
with patch.object(util.subprocess, "run", return_value=self._fail("not found")), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("missing:tag")
die.assert_called_once()
self.assertIn("missing:tag", die.call_args.args[0])
def test_dies_on_malformed_json(self):
with patch.object(util.subprocess, "run", return_value=self._ok("not-json {")), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
def test_dies_when_no_created_field(self):
payload = '[{"id": "sha256:abc123"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
self.assertIn("creation timestamp", die.call_args.args[0])
def test_dies_on_invalid_timestamp_format(self):
payload = '[{"created": "not-a-date"}]'
with patch.object(util.subprocess, "run", return_value=self._ok(payload)), \
patch.object(util, "die", side_effect=SystemExit("die")) as die:
with self.assertRaises(SystemExit):
util.image_created_at("some:tag")
die.assert_called_once()
if __name__ == "__main__": if __name__ == "__main__":
unittest.main() unittest.main()
+206
View File
@@ -0,0 +1,206 @@
"""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 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)
# ---------------------------------------------------------------------------
# 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()