feat(bottles): implement bottle factory abstraction per PRD 0003
test / run tests/run_tests.py (pull_request) Successful in 16s

Introduce claude_bottle/bottles/ with a Bottle Protocol and a
get_bottle_factory() that dispatches on CLAUDE_BOTTLE_PLATFORM
(default "docker"). Move every Docker-specific subprocess.run call
from cli/start.py, plus the orchestration of build, networks, the
pipelock sidecar, container launch, and per-container provisioning
(prompt, skills, ssh, .git), into create_docker_bottle.

Drop bottles[].runtime from the manifest schema. Auto-detect whether
gVisor is registered with the daemon and pass --runtime=runsc when it
is; the preflight shows the resolved runtime so the choice is visible.
Manifests still carrying 'runtime' get a clear error pointing at the
auto-detect behavior, rather than silent ignore.

Out of scope: cli/cleanup.py and cli/list.py still call docker
directly. They enumerate active bottles across the host, which is a
separate concern from "create a bottle" and is left for a follow-up
that introduces a list_active/cleanup primitive on the factory.
This commit is contained in:
2026-05-10 22:15:05 -04:00
parent d5c056f36e
commit d75cc9325f
8 changed files with 468 additions and 240 deletions
+4 -10
View File
@@ -45,9 +45,10 @@ like `cloudflare-dns.com` would have to be on the allowlist for the
agent to reach it at all. The container itself adds a layer between
the agent and the host, but the v1 design leans more on secret
minimization and egress allowlisting than on the container as a
hardened boundary. Linux hosts can opt into [gVisor](https://gvisor.dev/)
per bottle (see `runtime` in the manifest below) for a userspace
syscall barrier; the broader v2 discussion lives in
hardened boundary. On Linux hosts where [gVisor](https://gvisor.dev/)
is registered with Docker, claude-bottle auto-detects it and launches
every bottle under `runsc` for a userspace syscall barrier — no
manifest configuration required. The broader v2 discussion lives in
`docs/research/stronger-isolation-alternatives.md`.
The egress proxy and OAuth-token handling below are the load-bearing
@@ -76,13 +77,6 @@ project entries overriding home entries on key conflict).
{
"bottles": {
"gitea-dev": {
// Container runtime for the agent. Default "runc"; set to
// "runsc" on Linux hosts to launch the agent under gVisor for
// a userspace syscall barrier between the agent and the host
// kernel. claude-bottle verifies the runtime is registered with
// Docker before launch; gVisor is not available on macOS.
"runtime": "runsc",
"env": {
"GITEA_TOKEN": "?paste your Gitea API token",
"GITHUB_TOKEN": "${GH_PAT}",
-1
View File
@@ -13,7 +13,6 @@
},
"gitea-dev": {
"runtime": "runsc",
"env": {
"GITEA_TOKEN": "?paste your Gitea API token",
"GITHUB_TOKEN": "${GH_PAT}",
+54
View File
@@ -0,0 +1,54 @@
"""Per-platform bottle factories.
A bottle is a running, isolated environment with claude inside. Each
platform exposes a factory (currently only Docker) that owns the
end-to-end lifecycle: image build, container/sidecar launch, file
provisioning, and teardown.
Selection is driven by the CLAUDE_BOTTLE_PLATFORM env var (default
"docker"). Per PRD 0003 the manifest does not carry a platform field;
the host environment picks.
"""
from __future__ import annotations
import os
from contextlib import AbstractContextManager
from typing import Callable, Protocol
from ..log import die
from .docker import create_docker_bottle
class Bottle(Protocol):
"""Handle to a running bottle. Yielded by a factory's context manager.
`exec_claude` runs `claude` inside the bottle and blocks until the
session ends. `cp_in` copies a host path into the bottle. `close`
is an idempotent alias for context-manager teardown.
"""
name: str
def exec_claude(self, argv: list[str], *, tty: bool = True) -> int: ...
def cp_in(self, host_path: str, container_path: str) -> None: ...
def close(self) -> None: ...
BottleFactory = Callable[..., AbstractContextManager[Bottle]]
_FACTORIES: dict[str, BottleFactory] = {
"docker": create_docker_bottle,
}
def get_bottle_factory() -> BottleFactory:
"""Resolve the bottle factory for the active platform. Dies with a
pointer at the known platforms if CLAUDE_BOTTLE_PLATFORM names an
unimplemented one."""
name = os.environ.get("CLAUDE_BOTTLE_PLATFORM", "docker")
if name not in _FACTORIES:
known = ", ".join(sorted(_FACTORIES))
die(f"unknown CLAUDE_BOTTLE_PLATFORM={name!r}; known platforms: {known}")
return _FACTORIES[name]
+320
View File
@@ -0,0 +1,320 @@
"""Docker bottle factory.
`create_docker_bottle` owns the end-to-end Docker lifecycle:
1. Probe whether gVisor (`runsc`) is registered with the daemon.
2. Build the base image (and a per-cwd derived image if --cwd).
3. Create the per-agent internal + egress networks.
4. Boot the pipelock sidecar on both networks.
5. Launch the agent container, with `--runtime=runsc` iff available.
6. Copy the prompt, skills, SSH keys, and (optionally) .git into the
running container.
7. Yield a `Bottle` handle for `exec_claude` / `cp_in`.
8. Tear everything down (container, sidecar, both networks) on exit.
The Bottle Protocol lives in `claude_bottle.bottles.__init__`.
"""
from __future__ import annotations
import os
import subprocess
import sys
from contextlib import contextmanager
from dataclasses import dataclass
from pathlib import Path
from typing import Iterator
from .. import docker as docker_mod
from .. import network as network_mod
from .. import pipelock
from .. import skills as skills_mod
from .. import ssh as ssh_mod
from ..log import die, info
from ..manifest import Manifest
# --- Runtime detection -----------------------------------------------------
def runsc_available() -> bool:
"""Return True if the Docker daemon has the gVisor (`runsc`) runtime
registered. Called twice per `start`: once during the preflight to
render the runtime label, once inside the factory to set `--runtime`.
`docker info` is cheap; the duplication is not worth caching."""
r = subprocess.run(
["docker", "info", "--format", "{{json .Runtimes}}"],
capture_output=True,
text=True,
)
return r.returncode == 0 and "runsc" in r.stdout
def docker_runtime_label() -> str:
"""Human-readable label for the runtime that `create_docker_bottle`
would select right now. Shown in the y/N preflight."""
return "runsc (gVisor)" if runsc_available() else "runc (default)"
# --- Spec ------------------------------------------------------------------
@dataclass(frozen=True)
class DockerBottleSpec:
"""Host-side inputs assembled by the CLI before factory entry. Every
field is a value the factory consumes; nothing here is platform-
agnostic enough yet to lift into a shared spec (only Docker exists)."""
agent_name: str
slug: str
manifest: Manifest
container_name: str
container_name_pinned: bool
image: str
derived_image: str # "" -> no derived image
runtime_image: str # image to actually launch (derived or base)
user_cwd: str
copy_cwd_git: bool
stage_dir: Path
prompt_file: Path
env_file: Path
args_file: Path
pipelock_yaml_path: Path
pipelock_yaml_filename: str
forward_oauth_token: bool
# --- Bottle handle ---------------------------------------------------------
class _DockerBottle:
"""Concrete Bottle for Docker. Holds the resolved container name and
a teardown closure. Not exported — the factory yields it via the
Bottle Protocol."""
def __init__(self, container: str, teardown):
self.name = container
self._teardown = teardown
self._closed = False
def exec_claude(self, argv: list[str], *, tty: bool = True) -> int:
cmd = ["docker", "exec"]
if tty:
cmd.append("-it")
cmd.extend([self.name, "claude", *argv])
return subprocess.run(cmd).returncode
def cp_in(self, host_path: str, container_path: str) -> None:
subprocess.run(
["docker", "cp", host_path, f"{self.name}:{container_path}"],
stdout=subprocess.DEVNULL,
check=True,
)
def close(self) -> None:
if self._closed:
return
self._closed = True
self._teardown()
# --- Factory ---------------------------------------------------------------
# Where the repo root lives, for `docker build` context. Computed once.
_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent)
@contextmanager
def create_docker_bottle(spec: DockerBottleSpec) -> Iterator[_DockerBottle]:
"""Build, launch, and provision a Docker bottle. Teardown on exit."""
# Teardown bookkeeping. Each entry is populated as the matching
# resource comes up; teardown walks them in reverse, idempotently.
state: dict[str, str] = {
"container": "",
"pipelock": "",
"internal_network": "",
"egress_network": "",
}
def teardown() -> None:
try:
if state["container"] and docker_mod.container_exists(state["container"]):
subprocess.run(
["docker", "rm", "-f", state["container"]],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
state["container"] = ""
if state["pipelock"]:
pipelock.pipelock_stop(spec.slug)
state["pipelock"] = ""
if state["internal_network"]:
network_mod.network_remove(state["internal_network"])
state["internal_network"] = ""
if state["egress_network"]:
network_mod.network_remove(state["egress_network"])
state["egress_network"] = ""
except BaseException:
# Teardown must not raise; swallow so the caller's __exit__
# path can still propagate the original error.
pass
try:
use_runsc = runsc_available()
docker_mod.build_image(spec.image, _REPO_DIR)
if spec.derived_image:
docker_mod.build_image_with_cwd(spec.derived_image, spec.image, spec.user_cwd)
state["internal_network"] = network_mod.network_create_internal(spec.slug)
state["egress_network"] = network_mod.network_create_egress(spec.slug)
state["pipelock"] = pipelock.pipelock_start(
spec.slug,
state["internal_network"],
state["egress_network"],
spec.stage_dir,
spec.pipelock_yaml_filename,
)
container = _run_agent_container(spec, state["internal_network"], use_runsc)
state["container"] = container
_provision_container(spec, container)
bottle = _DockerBottle(container, teardown)
yield bottle
finally:
teardown()
# --- Internals -------------------------------------------------------------
def _run_agent_container(
spec: DockerBottleSpec,
internal_network: str,
use_runsc: bool,
) -> str:
"""Build the `docker run` argv and execute it, handling name-conflict
races by incrementing the suffix (unless the name was user-pinned).
Returns the resolved container name."""
proxy_url = pipelock.pipelock_proxy_url(spec.slug)
docker_args: list[str] = [
"--rm", "-d",
"--name", spec.container_name,
"--network", internal_network,
"-e", f"HTTPS_PROXY={proxy_url}",
"-e", f"HTTP_PROXY={proxy_url}",
"-e", "NO_PROXY=localhost,127.0.0.1",
]
if use_runsc:
docker_args.extend(["--runtime", "runsc"])
if spec.env_file.stat().st_size > 0:
docker_args.extend(["--env-file", str(spec.env_file)])
# ARGS_FILE pairs (-e, NAME) line-by-line.
args_lines = spec.args_file.read_text().splitlines()
i = 0
while i < len(args_lines):
flag = args_lines[i]
i += 1
if not flag:
continue
if i >= len(args_lines):
break
vname = args_lines[i]
i += 1
docker_args.extend([flag, vname])
if spec.forward_oauth_token:
os.environ["CLAUDE_CODE_OAUTH_TOKEN"] = os.environ["CLAUDE_BOTTLE_OAUTH_TOKEN"]
docker_args.extend(["-e", "CLAUDE_CODE_OAUTH_TOKEN"])
docker_args.extend([spec.runtime_image, "sleep", "infinity"])
info(f"starting container {spec.container_name} from {spec.runtime_image}")
container = spec.container_name
base_name = spec.container_name
suffix = 2
while True:
run_result = subprocess.run(
["docker", "run", *docker_args],
capture_output=True,
text=True,
)
if run_result.returncode == 0:
return container
err_text = run_result.stderr
if spec.container_name_pinned or "is already in use" not in err_text:
sys.stderr.write(err_text + "\n")
die(f"docker run failed for container '{container}'")
if suffix > 100:
die(
f"could not find a free container name after "
f"{base_name}-99 retries; clean up old containers"
)
container = f"{base_name}-{suffix}"
suffix += 1
name_idx = docker_args.index("--name") + 1
docker_args[name_idx] = container
info(f"name conflict; retrying as {container}")
def _provision_container(spec: DockerBottleSpec, container: str) -> None:
"""Copy prompt, skills, ssh keys, and (optionally) .git into the
running container, fixing up ownership/mode where the host UID
would otherwise leave files unreadable by the in-container `node`."""
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
container_prompt_path = f"{container_home}/.claude-bottle-prompt.txt"
subprocess.run(
["docker", "cp", str(spec.prompt_file), f"{container}:{container_prompt_path}"],
stdout=subprocess.DEVNULL,
check=True,
)
# `docker cp` preserves host UID; re-own/mode as root so node can
# read its own mode-600 prompt regardless of host UID.
subprocess.run(
["docker", "exec", "-u", "0", container, "chown", "node:node", container_prompt_path],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "exec", "-u", "0", container, "chmod", "600", container_prompt_path],
stdout=subprocess.DEVNULL,
check=True,
)
agent = spec.manifest.agents[spec.agent_name]
if agent.skills:
skills_mod.skills_copy_into(container, list(agent.skills))
bottle = spec.manifest.bottle_for(spec.agent_name)
if bottle.ssh:
proxy_host_port = pipelock.pipelock_proxy_host_port(spec.slug)
ssh_mod.ssh_setup(container, spec.stage_dir, proxy_host_port, bottle.ssh)
if spec.copy_cwd_git:
info(f"copying {spec.user_cwd}/.git -> {container}:/home/node/workspace/.git")
subprocess.run(
["docker", "cp", f"{spec.user_cwd}/.git", f"{container}:/home/node/workspace/.git"],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
[
"docker", "exec", "-u", "0", container,
"chown", "-R", "node:node", "/home/node/workspace/.git",
],
stdout=subprocess.DEVNULL,
check=True,
)
def container_prompt_path() -> str:
"""The path inside the container where the prompt file lands. Used
by start.py to pass `--append-system-prompt-file` to claude."""
container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
return f"{container_home}/.claude-bottle-prompt.txt"
+42 -169
View File
@@ -7,20 +7,24 @@ from __future__ import annotations
import argparse
import os
import shutil
import subprocess
import sys
import tempfile
from pathlib import Path
from .. import docker as docker_mod
from .. import network as network_mod
from .. import pipelock
from .. import skills as skills_mod
from .. import ssh as ssh_mod
from ..bottles import get_bottle_factory
from ..bottles.docker import (
DockerBottleSpec,
container_prompt_path,
docker_runtime_label,
)
from ..env_resolve import env_resolve
from ..log import die, info
from ..manifest import Manifest
from ._common import PROG, REPO_DIR, USER_CWD, read_tty_line
from ._common import PROG, USER_CWD, read_tty_line
def cmd_start(argv: list[str]) -> int:
@@ -86,10 +90,6 @@ def cmd_start(argv: list[str]) -> int:
if agent.skills:
skills_mod.skills_validate_all(list(agent.skills))
runtime = bottle.runtime
if runtime == "runsc":
docker_mod.require_runsc()
ssh_entries = bottle.ssh
if ssh_entries:
ssh_mod.ssh_validate_entries(ssh_entries)
@@ -106,31 +106,6 @@ def cmd_start(argv: list[str]) -> int:
prompt_file.write_text("")
prompt_file.chmod(0o600)
# cleanup state — populated as resources come up.
state: dict[str, str] = {
"container": "",
"pipelock": "",
"internal_network": "",
"egress_network": "",
}
def cleanup_all() -> None:
try:
if state["container"] and docker_mod.container_exists(state["container"]):
subprocess.run(
["docker", "rm", "-f", state["container"]],
stdout=subprocess.DEVNULL,
stderr=subprocess.DEVNULL,
)
if state["pipelock"]:
pipelock.pipelock_stop(slug)
if state["internal_network"]:
network_mod.network_remove(state["internal_network"])
if state["egress_network"]:
network_mod.network_remove(state["egress_network"])
finally:
shutil.rmtree(stage_dir, ignore_errors=True)
try:
pipelock.pipelock_write_yaml(manifest, bottle_name, pipelock_yaml)
allowlist_summary = pipelock.pipelock_allowlist_summary(manifest, bottle_name)
@@ -154,8 +129,8 @@ def cmd_start(argv: list[str]) -> int:
+ (", ".join(display_env_names) if display_env_names else "(none)")
)
info("skills : " + (" ".join(agent.skills) if agent.skills else "(none)"))
info(f"docker runtime : {docker_runtime_label()}")
info(f"bottle : {bottle_name}")
info(f" runtime : {runtime}{' (gVisor)' if runtime == 'runsc' else ''}")
if ssh_entries:
ssh_names = ", ".join(e.Host for e in ssh_entries)
info(f" ssh hosts : {ssh_names}")
@@ -171,7 +146,6 @@ def cmd_start(argv: list[str]) -> int:
if dry_run:
info("dry-run requested; not starting container.")
cleanup_all()
return 0
sys.stderr.write("claude-bottle: launch this agent? [y/N] ")
@@ -179,144 +153,43 @@ def cmd_start(argv: list[str]) -> int:
reply = read_tty_line()
if reply not in ("y", "Y", "yes", "YES"):
info("aborted by user")
cleanup_all()
return 0
# --- Build & launch ---
docker_mod.build_image(image, REPO_DIR)
if derived_image:
docker_mod.build_image_with_cwd(derived_image, image, USER_CWD)
state["internal_network"] = network_mod.network_create_internal(slug)
state["egress_network"] = network_mod.network_create_egress(slug)
state["pipelock"] = pipelock.pipelock_start(
slug,
state["internal_network"],
state["egress_network"],
stage_dir,
pipelock_yaml_filename,
spec = DockerBottleSpec(
agent_name=name,
slug=slug,
manifest=manifest,
container_name=container,
container_name_pinned=bool(pinned_container),
image=image,
derived_image=derived_image,
runtime_image=runtime_image,
user_cwd=USER_CWD,
copy_cwd_git=bool(args.cwd and Path(USER_CWD, ".git").is_dir()),
stage_dir=stage_dir,
prompt_file=prompt_file,
env_file=env_file,
args_file=args_file,
pipelock_yaml_path=pipelock_yaml,
pipelock_yaml_filename=pipelock_yaml_filename,
forward_oauth_token=forward_oauth_token,
)
proxy_url = pipelock.pipelock_proxy_url(slug)
docker_args: list[str] = [
"--rm", "-d",
"--name", container,
"--network", state["internal_network"],
"-e", f"HTTPS_PROXY={proxy_url}",
"-e", f"HTTP_PROXY={proxy_url}",
"-e", "NO_PROXY=localhost,127.0.0.1",
]
if runtime != "runc":
docker_args.extend(["--runtime", runtime])
if env_file.stat().st_size > 0:
docker_args.extend(["--env-file", str(env_file)])
# ARGS_FILE pairs (-e, NAME) line-by-line.
args_lines = args_file.read_text().splitlines()
i = 0
while i < len(args_lines):
flag = args_lines[i]
i += 1
if not flag:
continue
if i >= len(args_lines):
break
vname = args_lines[i]
i += 1
docker_args.extend([flag, vname])
if forward_oauth_token:
os.environ["CLAUDE_CODE_OAUTH_TOKEN"] = os.environ["CLAUDE_BOTTLE_OAUTH_TOKEN"]
docker_args.extend(["-e", "CLAUDE_CODE_OAUTH_TOKEN"])
docker_args.extend([runtime_image, "sleep", "infinity"])
info(f"starting container {container} from {runtime_image}")
# Retry-on-name-conflict loop to mirror the bash version.
while True:
full_argv = ["docker", "run", *docker_args]
run_result = subprocess.run(full_argv, capture_output=True, text=True)
if run_result.returncode == 0:
state["container"] = container
break
err_text = run_result.stderr
if pinned_container or "is already in use" not in err_text:
sys.stderr.write(err_text + "\n")
die(f"docker run failed for container '{container}'")
if suffix > 100:
die(
f"could not find a free container name after "
f"{default_container}-99 retries; clean up old containers"
factory = get_bottle_factory()
with factory(spec) as bottle_handle:
info(
"attaching interactive claude session "
"(Ctrl-D or 'exit' to leave; container will be removed)"
)
claude_args = ["--dangerously-skip-permissions"]
if args.remote_control:
claude_args.append("--remote-control")
if prompt_content:
claude_args.extend(
["--append-system-prompt-file", container_prompt_path()]
)
container = f"{default_container}-{suffix}"
suffix += 1
# Replace --name slot in docker_args.
name_idx = docker_args.index("--name") + 1
docker_args[name_idx] = container
info(f"name conflict; retrying as {container}")
container_prompt_path = (
os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
+ "/.claude-bottle-prompt.txt"
)
subprocess.run(
["docker", "cp", str(prompt_file), f"{container}:{container_prompt_path}"],
stdout=subprocess.DEVNULL,
check=True,
)
# `docker cp` preserves host UID; re-own/mode as root in the container
# so node can read its own mode-600 prompt regardless of host UID.
subprocess.run(
["docker", "exec", "-u", "0", container, "chown", "node:node", container_prompt_path],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "exec", "-u", "0", container, "chmod", "600", container_prompt_path],
stdout=subprocess.DEVNULL,
check=True,
)
if agent.skills:
skills_mod.skills_copy_into(container, list(agent.skills))
if ssh_entries:
proxy_host_port = pipelock.pipelock_proxy_host_port(slug)
ssh_mod.ssh_setup(container, stage_dir, proxy_host_port, ssh_entries)
if args.cwd and Path(USER_CWD, ".git").is_dir():
info(f"copying {USER_CWD}/.git -> {container}:/home/node/workspace/.git")
subprocess.run(
["docker", "cp", f"{USER_CWD}/.git", f"{container}:/home/node/workspace/.git"],
stdout=subprocess.DEVNULL,
check=True,
)
subprocess.run(
["docker", "exec", "-u", "0", container, "chown", "-R", "node:node", "/home/node/workspace/.git"],
stdout=subprocess.DEVNULL,
check=True,
)
info(
"attaching interactive claude session "
"(Ctrl-D or 'exit' to leave; container will be removed)"
)
claude_args = ["--dangerously-skip-permissions"]
if args.remote_control:
claude_args.append("--remote-control")
if prompt_content:
subprocess.run(
[
"docker", "exec", "-it", container, "claude",
*claude_args,
"--append-system-prompt-file", container_prompt_path,
]
)
else:
subprocess.run(
["docker", "exec", "-it", container, "claude", *claude_args]
)
info(f"session ended; container {container} will be removed")
return 0
bottle_handle.exec_claude(claude_args, tty=True)
info(f"session ended; container {bottle_handle.name} will be removed")
return 0
finally:
cleanup_all()
shutil.rmtree(stage_dir, ignore_errors=True)
-16
View File
@@ -19,22 +19,6 @@ def require_docker() -> None:
die("docker not found")
def require_runsc() -> None:
"""Fail with an install pointer if the `runsc` (gVisor) runtime is
not registered with the local Docker daemon. Called when a bottle
sets `runtime: "runsc"`."""
result = subprocess.run(
["docker", "info", "--format", "{{json .Runtimes}}"],
capture_output=True,
text=True,
)
if result.returncode != 0 or "runsc" not in result.stdout:
info("This bottle requested runtime 'runsc' but the gVisor runtime is not registered with Docker.")
info("Install gVisor and register it with the daemon: https://gvisor.dev/docs/user_guide/install/")
info("On macOS, gVisor is not available natively; remove 'runtime' from the bottle or run on Linux.")
die("runsc runtime not available")
def image_exists(ref: str) -> bool:
return _silent_run(["docker", "image", "inspect", ref]) == 0
+11 -23
View File
@@ -7,8 +7,7 @@ Schema (see CLAUDE.md "Intended design"):
"<bottle-name>": {
"env": { "<NAME>": <env-entry>, ... },
"ssh": [ <ssh-entry>, ... ],
"egress": { "allowlist": [ "<hostname>", ... ] },
"runtime": "runc" | "runsc"
"egress": { "allowlist": [ "<hostname>", ... ] }
}
},
"agents": {
@@ -33,15 +32,11 @@ import json
import os
from dataclasses import dataclass, field
from pathlib import Path
from typing import Literal, Mapping, cast
from typing import Mapping, cast
from .log import die
Runtime = Literal["runc", "runsc"]
_SUPPORTED_RUNTIMES: tuple[Runtime, ...] = ("runc", "runsc")
def _empty_str_dict() -> dict[str, str]:
return {}
@@ -116,12 +111,19 @@ class Bottle:
env: Mapping[str, str] = field(default_factory=_empty_str_dict)
ssh: tuple[SshEntry, ...] = ()
egress: BottleEgress = field(default_factory=BottleEgress)
runtime: Runtime = "runc"
@classmethod
def from_dict(cls, name: str, raw: object) -> "Bottle":
d = _as_json_object(raw, f"bottle '{name}'")
if "runtime" in d:
die(
f"bottle '{name}' has a 'runtime' field, which is no longer "
f"supported. gVisor (runsc) is now auto-detected when "
f"registered with Docker; remove the 'runtime' field from "
f"the bottle definition."
)
env: dict[str, str] = {}
env_raw = d.get("env")
if env_raw is not None:
@@ -152,21 +154,7 @@ class Bottle:
else BottleEgress()
)
runtime_raw = d.get("runtime")
runtime: Runtime
if runtime_raw is None:
runtime = "runc"
else:
if not isinstance(runtime_raw, str):
die(f"bottle '{name}' runtime must be a string (was {type(runtime_raw).__name__})")
if runtime_raw not in _SUPPORTED_RUNTIMES:
die(
f"bottle '{name}' runtime '{runtime_raw}' is not supported. "
f"Use one of: {', '.join(_SUPPORTED_RUNTIMES)}."
)
runtime = runtime_raw
return cls(env=env, ssh=ssh, egress=egress, runtime=runtime)
return cls(env=env, ssh=ssh, egress=egress)
@dataclass(frozen=True)
+37 -21
View File
@@ -1,16 +1,21 @@
"""Unit: bottle runtime — Manifest.from_json_obj defaults runtime to runc,
accepts runsc, and rejects unknown values, non-strings, and empty strings."""
"""Unit: bottle 'runtime' field is no longer supported (PRD 0003).
gVisor is now auto-detected by the Docker factory. A manifest carrying
the legacy 'runtime' field must fail loudly with a message pointing the
user at the auto-detect behavior, rather than silently ignoring."""
import io
import sys
import unittest
from claude_bottle.log import Die
from claude_bottle.manifest import Manifest
from claude_bottle.manifest import Bottle, Manifest
_ABSENT = object()
def _bottle(runtime_value: object) -> dict:
def _manifest(runtime_value: object) -> dict:
"""Build a minimal manifest JSON shape with one bottle whose runtime
field is set (or absent if `runtime_value is _ABSENT`)."""
bottle: dict = {}
@@ -22,30 +27,41 @@ def _bottle(runtime_value: object) -> dict:
}
class TestManifestBottleRuntime(unittest.TestCase):
def test_default_runc_when_absent(self):
m = Manifest.from_json_obj(_bottle(_ABSENT))
self.assertEqual("runc", m.bottles["dev"].runtime)
class TestManifestRuntimeRemoved(unittest.TestCase):
def test_loads_when_runtime_absent(self):
m = Manifest.from_json_obj(_manifest(_ABSENT))
self.assertIn("dev", m.bottles)
def test_explicit_runc(self):
m = Manifest.from_json_obj(_bottle("runc"))
self.assertEqual("runc", m.bottles["dev"].runtime)
def test_bottle_dataclass_has_no_runtime_attribute(self):
"""Structural check: the field has been removed from the dataclass."""
b = Bottle()
self.assertFalse(hasattr(b, "runtime"))
def test_explicit_runsc(self):
m = Manifest.from_json_obj(_bottle("runsc"))
self.assertEqual("runsc", m.bottles["dev"].runtime)
def test_rejects_runsc_value_with_helpful_message(self):
captured = io.StringIO()
old_stderr = sys.stderr
sys.stderr = captured
try:
with self.assertRaises(Die):
Manifest.from_json_obj(_manifest("runsc"))
finally:
sys.stderr = old_stderr
msg = captured.getvalue()
self.assertIn("'runtime'", msg, "error names the field")
self.assertIn("auto-detect", msg, "error points at the new behavior")
def test_rejects_unknown_runtime(self):
def test_rejects_runc_value(self):
with self.assertRaises(Die):
Manifest.from_json_obj(_bottle("kata-runtime"))
Manifest.from_json_obj(_manifest("runc"))
def test_rejects_unknown_value(self):
with self.assertRaises(Die):
Manifest.from_json_obj(_manifest("kata-runtime"))
def test_rejects_non_string(self):
"""Any presence of the field is an error; type is not consulted."""
with self.assertRaises(Die):
Manifest.from_json_obj(_bottle(42))
def test_rejects_empty_string(self):
with self.assertRaises(Die):
Manifest.from_json_obj(_bottle(""))
Manifest.from_json_obj(_manifest(42))
if __name__ == "__main__":