Merge pull request 'PRD 0003: Bottle Backend abstraction' (#5) from add-bottle-factory-abstraction into main

README.md

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

claude-bottle.example.json

@@ -13,7 +13,6 @@
     },
 
     "gitea-dev": {
-      "runtime": "runsc",
       "env": {
         "GITEA_TOKEN":     "?paste your Gitea API token",
         "GITHUB_TOKEN":    "${GH_PAT}",

claude_bottle/backend/__init__.py

@@ -0,0 +1,211 @@
+"""Per-backend bottle factories.
+
+A bottle is a running, isolated environment with claude inside. Each
+backend exposes five methods:
+
+  prepare(spec, stage_dir=...) -> BottlePlan
+      Resolves names, validates host-side prerequisites, and writes
+      scratch files. No remote/runtime resources are created yet.
+      Safe to call before the y/N preflight.
+
+  launch(plan) -> ContextManager[Bottle]
+      Brings up the container (or VM, or remote machine), provisions
+      it, yields a Bottle handle, and tears everything down on exit.
+
+  prepare_cleanup() -> BottleCleanupPlan
+      Enumerates orphaned resources left behind by previous bottles
+      (containers, networks, ...). Idempotent; no side effects.
+
+  cleanup(plan) -> None
+      Actually removes everything described by the cleanup plan.
+
+  list_active() -> None
+      Print every currently-running bottle on this backend to stderr.
+
+Selection is driven by CLAUDE_BOTTLE_BACKEND (default "docker"). Per
+PRD 0003 the manifest does not carry a backend field; the host
+environment picks.
+"""
+
+from __future__ import annotations
+
+import os
+from abc import ABC, abstractmethod
+from contextlib import AbstractContextManager
+from dataclasses import dataclass
+from pathlib import Path
+
+from ..log import die
+from ..manifest import Manifest
+
+
+@dataclass(frozen=True)
+class BottleSpec:
+    """CLI-supplied intent. Backend-agnostic — each backend's prepare
+    step consumes it and produces its own backend-specific plan.
+    Resolved values (image names, container name, scratch paths, runsc
+    availability) live on the plan, not the spec."""
+
+    manifest: Manifest
+    agent_name: str
+    copy_cwd: bool
+    user_cwd: str
+    forward_oauth_token: bool
+
+
+@dataclass(frozen=True)
+class BottlePlan(ABC):
+    """Base output of a backend's prepare step. Concrete subclasses
+    (e.g. DockerBottlePlan) add backend-specific resolved fields and
+    implement `print`."""
+
+    spec: BottleSpec
+    stage_dir: Path
+
+    @abstractmethod
+    def print(self, *, remote_control: bool) -> None:
+        """Render the y/N preflight summary to stderr."""
+
+
+@dataclass(frozen=True)
+class BottleCleanupPlan(ABC):
+    """Base output of a backend's prepare_cleanup step. Concrete
+    subclasses (e.g. DockerBottleCleanupPlan) carry backend-specific
+    lists of resources to be removed and implement `print` + `empty`."""
+
+    @abstractmethod
+    def print(self) -> None:
+        """Render the cleanup y/N summary to stderr."""
+
+    @property
+    @abstractmethod
+    def empty(self) -> bool:
+        """True iff there is nothing to clean up; the CLI uses this to
+        short-circuit before showing the y/N."""
+
+
+class Bottle(ABC):
+    """Handle to a running bottle. Yielded by a backend's launch step.
+
+    `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
+
+    @abstractmethod
+    def exec_claude(self, argv: list[str], *, tty: bool = True) -> int: ...
+
+    @abstractmethod
+    def cp_in(self, host_path: str, container_path: str) -> None: ...
+
+    @abstractmethod
+    def close(self) -> None: ...
+
+
+
+
+class BottleBackend(ABC):
+    """Abstract base for selectable bottle backends. Concrete subclasses
+    (e.g. DockerBottleBackend) own their own prepare/launch impls.
+    Symmetric with the BottlePlan → DockerBottlePlan hierarchy."""
+
+    name: str
+
+    @abstractmethod
+    def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> BottlePlan:
+        """Resolve names, validate host-side prerequisites, write
+        scratch files. No remote/runtime resources created yet."""
+
+    @abstractmethod
+    def launch(self, plan: BottlePlan) -> AbstractContextManager[Bottle]:
+        """Build/run the bottle and yield a handle; tear down on exit."""
+
+    def provision(self, plan: BottlePlan, target: str) -> str | None:
+        """Copy host-side files (prompt, skills, SSH keys, .git) into
+        the running bottle. Called from `launch` after the container/
+        machine is up. `target` identifies the running instance in
+        backend-specific terms (Docker: resolved container name; fly:
+        machine id). Returns the in-container prompt path if a prompt
+        was provisioned, else None — the Bottle handle uses it to
+        decide whether to add --append-system-prompt-file to claude's
+        argv.
+
+        Default orchestration: prompt → skills → ssh → git. Subclasses
+        typically don't override this; they implement the four
+        sub-methods below."""
+        prompt_path = self.provision_prompt(plan, target)
+        self.provision_skills(plan, target)
+        self.provision_ssh(plan, target)
+        self.provision_git(plan, target)
+        return prompt_path
+
+    @abstractmethod
+    def provision_prompt(self, plan: BottlePlan, target: str) -> str | None:
+        """Copy the prompt file into the running bottle. Returns the
+        in-container path iff the agent has a non-empty prompt;
+        callers use the return value to decide whether to add
+        --append-system-prompt-file to claude's argv."""
+
+    @abstractmethod
+    def provision_skills(self, plan: BottlePlan, target: str) -> None:
+        """Copy the agent's named skills from the host into the
+        running bottle. No-op when the agent has no skills."""
+
+    @abstractmethod
+    def provision_ssh(self, plan: BottlePlan, target: str) -> None:
+        """Set up SSH in the running bottle (config, agent, keys)
+        so the bottle can reach the manifest's declared SSH hosts.
+        No-op when the bottle has no SSH entries."""
+
+    @abstractmethod
+    def provision_git(self, plan: BottlePlan, target: str) -> None:
+        """Copy the host's cwd `.git` directory into the running
+        bottle if the user requested --cwd. No-op otherwise."""
+
+    @abstractmethod
+    def prepare_cleanup(self) -> BottleCleanupPlan:
+        """Enumerate orphaned resources from previous bottles. No side
+        effects; safe to call before the y/N."""
+
+    @abstractmethod
+    def cleanup(self, plan: BottleCleanupPlan) -> None:
+        """Remove everything described by the cleanup plan."""
+
+    @abstractmethod
+    def list_active(self) -> None:
+        """Print every currently-running bottle on this backend to
+        stderr (name + status)."""
+
+
+# Import concrete backend classes AFTER the base types are defined, so
+# each backend module can pull BottleSpec / BottlePlan / BottleBackend
+# via `from . import ...` without hitting a partially-initialized module.
+from .docker import DockerBottleBackend  # noqa: E402
+
+
+_BACKENDS: dict[str, BottleBackend] = {
+    "docker": DockerBottleBackend(),
+}
+
+
+def get_bottle_backend() -> BottleBackend:
+    """Resolve the bottle backend for the active environment. Dies with
+    a pointer at the known backends if CLAUDE_BOTTLE_BACKEND names an
+    unimplemented one."""
+    name = os.environ.get("CLAUDE_BOTTLE_BACKEND", "docker")
+    if name not in _BACKENDS:
+        known = ", ".join(sorted(_BACKENDS))
+        die(f"unknown CLAUDE_BOTTLE_BACKEND={name!r}; known backends: {known}")
+    return _BACKENDS[name]
+
+
+__all__ = [
+    "Bottle",
+    "BottleBackend",
+    "BottleCleanupPlan",
+    "BottlePlan",
+    "BottleSpec",
+    "get_bottle_backend",
+]

claude_bottle/backend/docker/__init__.py

@@ -0,0 +1,29 @@
+"""Docker bottle backend.
+
+The bulk of the implementation lives in sibling modules:
+
+  - util:                 thin Docker subprocess wrappers
+  - network:              Docker network plumbing
+  - bottle_plan:          DockerBottlePlan
+  - bottle_cleanup_plan:  DockerBottleCleanupPlan
+  - bottle:               DockerBottle handle
+  - backend:              DockerBottleBackend
+
+This file only re-exports the public names so
+`from claude_bottle.backend.docker import DockerBottleBackend` keeps
+working.
+"""
+
+from __future__ import annotations
+
+from .backend import DockerBottleBackend
+from .bottle import DockerBottle
+from .bottle_cleanup_plan import DockerBottleCleanupPlan
+from .bottle_plan import DockerBottlePlan
+
+__all__ = [
+    "DockerBottle",
+    "DockerBottleBackend",
+    "DockerBottleCleanupPlan",
+    "DockerBottlePlan",
+]

claude_bottle/backend/docker/backend.py

@@ -0,0 +1,674 @@
+"""DockerBottleBackend — the Docker implementation of BottleBackend.
+
+Methods:
+  .prepare(spec, stage_dir=...) -> DockerBottlePlan
+  .launch(plan) -> ContextManager[DockerBottle]
+  .prepare_cleanup() -> DockerBottleCleanupPlan
+  .cleanup(plan) -> None
+  .list_active() -> None
+"""
+
+from __future__ import annotations
+
+import dataclasses
+import os
+import subprocess
+import sys
+from contextlib import contextmanager
+from pathlib import Path
+from typing import Iterator, Sequence
+
+from ... import pipelock
+from ...env import ResolvedEnv, resolve_env
+from ...log import die, info
+from ...manifest import SshEntry
+from ...util import expand_tilde
+from .. import BottleBackend, BottleCleanupPlan, BottlePlan, BottleSpec
+from ..util import host_skill_dir
+from . import network as network_mod
+from . import util as docker_mod
+from .bottle import DockerBottle
+from .bottle_cleanup_plan import DockerBottleCleanupPlan
+from .bottle_plan import DockerBottlePlan
+from .pipelock import (
+    DockerPipelockProxy,
+    pipelock_proxy_host_port,
+    pipelock_proxy_url,
+)
+
+
+# Where the repo root lives, for `docker build` context. Computed once.
+_REPO_DIR = str(Path(__file__).resolve().parent.parent.parent.parent)
+
+
+class DockerBottleBackend(BottleBackend):
+    """Docker backend implementation. Selected by CLAUDE_BOTTLE_BACKEND
+    (default)."""
+
+    name = "docker"
+    _proxy: DockerPipelockProxy = DockerPipelockProxy()
+
+    def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
+        """Resolve names, validate, write scratch files. No Docker
+        resources are created; the only side effects are host-side
+        files under stage_dir and a probe of `docker info`."""
+        docker_mod.require_docker()
+
+        manifest = spec.manifest
+        manifest.require_agent(spec.agent_name)
+        agent = manifest.agents[spec.agent_name]
+        bottle = manifest.bottle_for(spec.agent_name)
+
+        slug = docker_mod.slugify(spec.agent_name)
+
+        image = os.environ.get("CLAUDE_BOTTLE_IMAGE", "claude-bottle:latest")
+        derived_image = ""
+        runtime_image = image
+        if spec.copy_cwd:
+            derived_image = os.environ.get(
+                "CLAUDE_BOTTLE_DERIVED_IMAGE", f"claude-bottle:cwd-{slug}"
+            )
+            runtime_image = derived_image
+
+        default_container = f"claude-bottle-{slug}"
+        pinned_container = os.environ.get("CLAUDE_BOTTLE_CONTAINER", "")
+        container_name = pinned_container or default_container
+        container_name_pinned = bool(pinned_container)
+        suffix = 2
+        if container_name_pinned:
+            if docker_mod.container_exists(container_name):
+                die(
+                    f"container '{container_name}' already exists "
+                    f"(pinned via CLAUDE_BOTTLE_CONTAINER). "
+                    f"Remove it with 'docker rm -f {container_name}' or unset the override."
+                )
+        else:
+            while docker_mod.container_exists(container_name):
+                container_name = f"{default_container}-{suffix}"
+                suffix += 1
+                if suffix > 100:
+                    die(
+                        f"could not find a free container name after "
+                        f"{default_container}-99; clean up old containers with "
+                        f"'docker rm -f <name>'"
+                    )
+
+        if agent.skills:
+            self.validate_skills(list(agent.skills))
+        if bottle.ssh:
+            self.validate_ssh_entries(bottle.ssh)
+
+        env_file = stage_dir / "agent.env"
+        args_file = stage_dir / "docker-args"
+        prompt_file = stage_dir / "prompt.txt"
+        prompt_file.write_text("")
+        prompt_file.chmod(0o600)
+
+        proxy_plan = self.prepare_proxy(spec, stage_dir)
+        resolved = resolve_env(manifest, spec.agent_name)
+        self._write_env_files(resolved, env_file, args_file)
+        prompt_file.write_text(agent.prompt)
+
+        allowlist_summary = pipelock.pipelock_allowlist_summary(bottle)
+        use_runsc = docker_mod.runsc_available()
+
+        return DockerBottlePlan(
+            spec=spec,
+            stage_dir=stage_dir,
+            slug=slug,
+            container_name=container_name,
+            container_name_pinned=container_name_pinned,
+            image=image,
+            derived_image=derived_image,
+            runtime_image=runtime_image,
+            env_file=env_file,
+            args_file=args_file,
+            prompt_file=prompt_file,
+            proxy_plan=proxy_plan,
+            allowlist_summary=allowlist_summary,
+            use_runsc=use_runsc,
+        )
+
+    def _write_env_files(
+        self, resolved: ResolvedEnv, env_file: Path, args_file: Path
+    ) -> None:
+        """Serialize a ResolvedEnv into the two on-disk formats the launch
+        step consumes: `--env-file` syntax for literals (NAME=VALUE per
+        line) and a paired `-e\\nNAME\\n` stream for forwarded names.
+        Both files are created here (mode 600 on the literals file,
+        which may carry sensitive verbatim values from the manifest)."""
+        env_lines: list[str] = []
+        for name, value in resolved.literals.items():
+            if "\n" in value:
+                die(
+                    f"env entry {name} (literal) contains a newline; "
+                    f"docker --env-file cannot represent multi-line values."
+                )
+            env_lines.append(f"{name}={value}")
+        env_file.write_text("\n".join(env_lines) + ("\n" if env_lines else ""))
+        env_file.chmod(0o600)
+
+        args_lines = [f"-e\n{name}" for name in resolved.forwarded]
+        args_file.write_text("\n".join(args_lines) + ("\n" if args_lines else ""))
+
+    def prepare_proxy(self, spec: BottleSpec, stage_dir: Path) -> pipelock.PipelockProxyPlan:
+        """Decide where the pipelock yaml lives in `stage_dir`, delegate
+        to PipelockProxy to write it, and return the resolved
+        PipelockProxyPlan for the launch step to consume. Stage-only:
+        no Docker resources created yet."""
+        yaml_path = stage_dir / "pipelock.yaml"
+        bottle = spec.manifest.bottle_for(spec.agent_name)
+        slug = docker_mod.slugify(spec.agent_name)
+        return self._proxy.prepare(bottle, slug, yaml_path)
+
+    @contextmanager
+    def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]:
+        """Build, launch, and provision a Docker bottle. Teardown on exit."""
+        assert isinstance(plan, DockerBottlePlan), (
+            f"DockerBottleBackend.launch expects DockerBottlePlan, "
+            f"got {type(plan).__name__}"
+        )
+
+        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"]:
+                    self._proxy.stop(state["pipelock"])
+                    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:
+            docker_mod.build_image(plan.image, _REPO_DIR)
+            if plan.derived_image:
+                docker_mod.build_image_with_cwd(
+                    plan.derived_image, plan.image, plan.spec.user_cwd
+                )
+
+            state["internal_network"] = network_mod.network_create_internal(plan.slug)
+            state["egress_network"] = network_mod.network_create_egress(plan.slug)
+            proxy_plan = dataclasses.replace(
+                plan.proxy_plan,
+                internal_network=state["internal_network"],
+                egress_network=state["egress_network"],
+            )
+            state["pipelock"] = self._proxy.start(proxy_plan)
+
+            container = self._run_agent_container(plan, state["internal_network"])
+            state["container"] = container
+
+            prompt_path = self.provision(plan, container)
+
+            bottle = DockerBottle(container, teardown, prompt_path)
+            yield bottle
+        finally:
+            teardown()
+
+    def _run_agent_container(self, plan: DockerBottlePlan, internal_network: str) -> 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_proxy_url(plan.slug)
+        docker_args: list[str] = [
+            "--rm", "-d",
+            "--name", plan.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 plan.use_runsc:
+            docker_args.extend(["--runtime", "runsc"])
+        if plan.env_file.stat().st_size > 0:
+            docker_args.extend(["--env-file", str(plan.env_file)])
+
+        # ARGS_FILE pairs (-e, NAME) line-by-line.
+        args_lines = plan.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 plan.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([plan.runtime_image, "sleep", "infinity"])
+
+        info(f"starting container {plan.container_name} from {plan.runtime_image}")
+
+        container = plan.container_name
+        base_name = plan.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 plan.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_prompt(self, plan: BottlePlan, target: str) -> str | None:
+        """Copy the prompt file into the container, fix ownership/mode.
+        Returns the in-container path if the agent has a non-empty
+        prompt (drives --append-system-prompt-file), else None. The
+        file is copied either way so the path always exists."""
+        assert isinstance(plan, DockerBottlePlan)
+        container = target
+        container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
+        in_container_prompt_path = f"{container_home}/.claude-bottle-prompt.txt"
+
+        subprocess.run(
+            ["docker", "cp", str(plan.prompt_file), f"{container}:{in_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", in_container_prompt_path],
+            stdout=subprocess.DEVNULL,
+            check=True,
+        )
+        subprocess.run(
+            ["docker", "exec", "-u", "0", container, "chmod", "600", in_container_prompt_path],
+            stdout=subprocess.DEVNULL,
+            check=True,
+        )
+
+        agent = plan.spec.manifest.agents[plan.spec.agent_name]
+        return in_container_prompt_path if agent.prompt else None
+
+    def validate_skills(self, skills: list[str]) -> None:
+        """Fail loudly if any named skill is missing from the host's
+        ~/.claude/skills/. Called from `prepare` before the y/N so the
+        user doesn't get a launch prompt for a plan that's already
+        known to break."""
+        for name in skills:
+            path = host_skill_dir(name)
+            if not os.path.isdir(path):
+                die(
+                    f"skill '{name}' not found on host at {path}. "
+                    f"Create it under ~/.claude/skills/, then re-run."
+                )
+
+    def provision_skills(self, plan: BottlePlan, target: str) -> None:
+        """Copy each of the agent's named skills from the host's
+        ~/.claude/skills/<name>/ into the container's equivalent path.
+        For each skill: ensure parent dir, wipe any prior copy, then
+        `docker cp <host>/. <container>:<dst>/` so the contents are
+        copied into a freshly-created destination dir. No-op when the
+        agent has no skills."""
+        assert isinstance(plan, DockerBottlePlan)
+        agent = plan.spec.manifest.agents[plan.spec.agent_name]
+        if not agent.skills:
+            return
+
+        container = target
+        container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
+        skills_dir = os.environ.get(
+            "CLAUDE_BOTTLE_CONTAINER_SKILLS_DIR", f"{container_home}/.claude/skills"
+        )
+
+        subprocess.run(
+            ["docker", "exec", container, "mkdir", "-p", skills_dir],
+            stdout=subprocess.DEVNULL,
+            check=True,
+        )
+
+        for n in agent.skills:
+            src = host_skill_dir(n)
+            if not os.path.isdir(src):
+                die(f"skill '{n}' disappeared from host between validation and copy at {src}.")
+            dst = f"{skills_dir}/{n}"
+            info(f"copying skill {n} into {container}:{dst}")
+            subprocess.run(
+                ["docker", "exec", container, "rm", "-rf", dst],
+                stdout=subprocess.DEVNULL,
+                check=True,
+            )
+            subprocess.run(
+                ["docker", "exec", container, "mkdir", "-p", dst],
+                stdout=subprocess.DEVNULL,
+                check=True,
+            )
+            subprocess.run(
+                ["docker", "cp", f"{src}/.", f"{container}:{dst}/"],
+                stdout=subprocess.DEVNULL,
+                check=True,
+            )
+
+    def validate_ssh_entries(self, entries: Sequence[SshEntry]) -> None:
+        """Each entry's IdentityFile must exist on the host (after
+        expanding leading ~). Host and IdentityFile shape are already
+        enforced by Manifest validation. Called from `prepare` before
+        the y/N so the user doesn't get prompted for a plan with a
+        missing key."""
+        for entry in entries:
+            key = expand_tilde(entry.IdentityFile)
+            if not os.path.isfile(key):
+                die(f"ssh key file not found for host '{entry.Host}': {key}")
+
+    def provision_ssh(self, plan: BottlePlan, target: str) -> None:
+        """Set up SSH in the container so node can authenticate using
+        each entry's key without the key file being readable by node.
+        No-op when the bottle has no SSH entries.
+
+        Isolation strategy:
+          - Keys live at /root/.claude-bottle-keys/ (mode 700,
+            root-owned). /root is mode 700 in node:22-slim, so node
+            (uid 1000) can't even traverse in.
+          - ssh-agent runs as root, listening on
+            /run/claude-bottle-agent.sock. Each key is loaded with
+            ssh-add, then deleted; the bytes now live only in the
+            agent process's memory.
+          - ssh-agent's SO_PEERCRED-based UID match rejects every
+            connection whose peer euid is neither 0 nor the agent's.
+            To bridge that, a root-owned socat forwarder listens on
+            /run/claude-bottle-agent-public.sock (mode 666) and
+            proxies bytes to the real agent socket.
+          - node can't ptrace root-owned agent or socat, so
+            /proc/<pid>/mem is off-limits and key bytes never leave
+            root-owned memory.
+          - ~/.ssh/config in node's home points each Host at the
+            public socket via IdentityAgent.
+
+        Why an in-container agent (not bind-mounted from host):
+        Docker Desktop on macOS does not forward Unix-domain socket
+        connect() across the VM boundary — connect() returns
+        ENOTSUP. Running ssh-agent inside the container sidesteps
+        that entirely.
+
+        Limitation: keys must be passphrase-less. ssh-add prompts on
+        /dev/tty for passphrases, but our docker exec has no TTY."""
+        assert isinstance(plan, DockerBottlePlan)
+        bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
+        if not bottle.ssh:
+            return
+
+        container = target
+        proxy_host_port = pipelock_proxy_host_port(plan.slug)
+        container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
+        container_ssh = f"{container_home}/.ssh"
+        agent_socket = "/run/claude-bottle-agent.sock"
+        public_socket = "/run/claude-bottle-agent-public.sock"
+        keys_dir = "/root/.claude-bottle-keys"
+
+        # ~/.ssh for node (700, owned by node).
+        docker_mod.docker_exec_root(container, ["mkdir", "-p", container_ssh])
+        docker_mod.docker_exec_root(container, ["chown", "node:node", container_ssh])
+        docker_mod.docker_exec_root(container, ["chmod", "700", container_ssh])
+
+        # /root/.claude-bottle-keys for root (700, root-owned).
+        docker_mod.docker_exec_root(container, ["mkdir", "-p", keys_dir])
+        docker_mod.docker_exec_root(container, ["chown", "root:root", keys_dir])
+        docker_mod.docker_exec_root(container, ["chmod", "700", keys_dir])
+
+        config_file = plan.stage_dir / "ssh_config"
+        known_hosts_file = plan.stage_dir / "ssh_known_hosts"
+        config_file.write_text("")
+        config_file.chmod(0o600)
+        known_hosts_file.write_text("")
+        known_hosts_file.chmod(0o600)
+
+        proxy_host, _, proxy_port = proxy_host_port.partition(":")
+
+        container_key_paths: list[str] = []
+        for entry in bottle.ssh:
+            name = entry.Host
+            key = expand_tilde(entry.IdentityFile)
+            hostname = entry.Hostname
+            user = entry.User
+            port = entry.Port
+            known_host_key = entry.KnownHostKey
+
+            key_basename = os.path.basename(key)
+            container_key_path = f"{keys_dir}/{key_basename}"
+
+            info(f"copying ssh key for '{name}' -> {container} (root-only staging)")
+            subprocess.run(
+                ["docker", "cp", key, f"{container}:{container_key_path}"],
+                stdout=subprocess.DEVNULL,
+                check=True,
+            )
+            docker_mod.docker_exec_root(container, ["chown", "root:root", container_key_path])
+            docker_mod.docker_exec_root(container, ["chmod", "600", container_key_path])
+
+            container_key_paths.append(container_key_path)
+
+            # ProxyCommand tunnels SSH through pipelock via HTTP
+            # CONNECT. %h / %p expand to this block's HostName /
+            # Port. socat's PROXY: mode does CONNECT host:port to
+            # the proxy.
+            block = (
+                f"Host {name}\n"
+                f"  HostName {hostname}\n"
+                f"  User {user}\n"
+                f"  Port {port}\n"
+                f"  IdentityAgent {public_socket}\n"
+                f"  ProxyCommand socat - PROXY:{proxy_host}:%h:%p,proxyport={proxy_port}\n"
+                f"\n"
+            )
+            with config_file.open("a") as f:
+                f.write(block)
+
+            if known_host_key:
+                entries_to_write: list[str] = []
+                if port == "22":
+                    entries_to_write.append(f"{name} {known_host_key}\n")
+                    if hostname != name:
+                        entries_to_write.append(f"{hostname} {known_host_key}\n")
+                else:
+                    entries_to_write.append(f"[{name}]:{port} {known_host_key}\n")
+                    if hostname != name:
+                        entries_to_write.append(f"[{hostname}]:{port} {known_host_key}\n")
+                with known_hosts_file.open("a") as f:
+                    for e in entries_to_write:
+                        f.write(e)
+
+        # Boot the agent, load each key, delete the key files, then
+        # start the root-owned socat forwarder. One docker exec so the
+        # whole sequence is atomic.
+        info(f"starting in-container ssh-agent at {agent_socket} (forwarded via {public_socket})")
+        setup_lines = [
+            "set -eu",
+            f"ssh-agent -a {agent_socket} >/dev/null",
+        ]
+        for kp in container_key_paths:
+            setup_lines.append(f"SSH_AUTH_SOCK={agent_socket} ssh-add {kp}")
+            setup_lines.append(f"rm -f {kp}")
+        setup_lines.append(f"rmdir {keys_dir} 2>/dev/null || true")
+        # Forwarder: socat (uid 0) connects to the agent on node's behalf.
+        setup_lines.append(
+            f"nohup socat UNIX-LISTEN:{public_socket},fork,reuseaddr,mode=666 "
+            f"UNIX-CONNECT:{agent_socket} </dev/null >/dev/null 2>&1 &"
+        )
+        # Wait briefly for the forwarder to bind.
+        setup_lines.extend([
+            "i=0",
+            "while [ $i -lt 20 ]; do",
+            f"  [ -S {public_socket} ] && break",
+            "  i=$((i + 1))",
+            "  sleep 0.1",
+            "done",
+            f"[ -S {public_socket} ] || {{ echo 'claude-bottle: socat forwarder failed to bind {public_socket}' >&2; exit 1; }}",
+        ])
+        setup_script = "\n".join(setup_lines) + "\n"
+        subprocess.run(
+            ["docker", "exec", "-u", "0", container, "sh", "-c", setup_script],
+            check=True,
+        )
+
+        info(f"writing {container_ssh}/config")
+        subprocess.run(
+            ["docker", "cp", str(config_file), f"{container}:{container_ssh}/config"],
+            stdout=subprocess.DEVNULL,
+            check=True,
+        )
+        docker_mod.docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/config"])
+        docker_mod.docker_exec_root(container, ["chmod", "600", f"{container_ssh}/config"])
+
+        if known_hosts_file.stat().st_size > 0:
+            info(f"writing {container_ssh}/known_hosts")
+            subprocess.run(
+                ["docker", "cp", str(known_hosts_file), f"{container}:{container_ssh}/known_hosts"],
+                stdout=subprocess.DEVNULL,
+                check=True,
+            )
+            docker_mod.docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/known_hosts"])
+            docker_mod.docker_exec_root(container, ["chmod", "600", f"{container_ssh}/known_hosts"])
+
+    def provision_git(self, plan: BottlePlan, target: str) -> None:
+        """If --cwd was set and the host cwd has a .git directory, copy
+        it into /home/node/workspace/.git and fix ownership. No-op
+        otherwise."""
+        assert isinstance(plan, DockerBottlePlan)
+        if not (plan.spec.copy_cwd and Path(plan.spec.user_cwd, ".git").is_dir()):
+            return
+        container = target
+        info(f"copying {plan.spec.user_cwd}/.git -> {container}:/home/node/workspace/.git")
+        subprocess.run(
+            ["docker", "cp", f"{plan.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,
+        )
+
+    # --- Cleanup ---
+
+    def prepare_cleanup(self) -> DockerBottleCleanupPlan:
+        """Enumerate all claude-bottle-prefixed containers (running or
+        stopped) and networks. No removals — caller confirms first."""
+        docker_mod.require_docker()
+
+        # `docker ps -a --filter name=...` uses regex matching; anchor at
+        # the start so we don't pick up containers that merely contain
+        # "claude-bottle-" mid-name.
+        cr = subprocess.run(
+            [
+                "docker", "ps", "-a",
+                "--filter", "name=^claude-bottle-",
+                "--format", "{{.Names}}",
+            ],
+            capture_output=True,
+            text=True,
+        )
+        containers = tuple(sorted(
+            line for line in (cr.stdout or "").splitlines() if line
+        ))
+
+        # `docker network ls --filter name=...` uses substring matching.
+        # "claude-bottle-" is specific enough that false positives are
+        # not a concern.
+        nr = subprocess.run(
+            [
+                "docker", "network", "ls",
+                "--filter", "name=claude-bottle-",
+                "--format", "{{.Name}}",
+            ],
+            capture_output=True,
+            text=True,
+        )
+        networks = tuple(sorted(
+            line for line in (nr.stdout or "").splitlines() if line
+        ))
+
+        return DockerBottleCleanupPlan(containers=containers, networks=networks)
+
+    def cleanup(self, plan: BottleCleanupPlan) -> None:
+        """Remove the containers and networks listed in the plan.
+        Containers first; networks would refuse to delete while
+        containers are still attached."""
+        assert isinstance(plan, DockerBottleCleanupPlan), (
+            f"DockerBottleBackend.cleanup expects DockerBottleCleanupPlan, "
+            f"got {type(plan).__name__}"
+        )
+        for name in plan.containers:
+            info(f"removing container {name}")
+            subprocess.run(
+                ["docker", "rm", "-f", name],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+        for name in plan.networks:
+            info(f"removing network {name}")
+            subprocess.run(
+                ["docker", "network", "rm", name],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            )
+
+    # --- List ---
+
+    def list_active(self) -> None:
+        """Print all running claude-bottle containers (name + status).
+        Prints a single-line banner if there are none."""
+        docker_mod.require_docker()
+        result = subprocess.run(
+            [
+                "docker", "ps",
+                "--filter", "name=^claude-bottle-",
+                "--format", "{{.Names}}\t{{.Status}}",
+            ],
+            capture_output=True,
+            text=True,
+        )
+        containers = (result.stdout or "").strip()
+        if not containers:
+            info("no active claude-bottle containers")
+            return
+        print()
+        for line in containers.splitlines():
+            name, _, status = line.partition("\t")
+            info(f"container: {name}  status: {status}")
+        print()

claude_bottle/backend/docker/bottle.py

@@ -0,0 +1,46 @@
+"""DockerBottle — concrete Bottle handle yielded by
+DockerBottleBackend.launch.
+
+Holds the container name plus the in-container prompt path so
+exec_claude can transparently add --append-system-prompt-file when a
+prompt was provisioned.
+"""
+
+from __future__ import annotations
+
+import subprocess
+
+from .. import Bottle
+
+
+class DockerBottle(Bottle):
+    """Concrete Bottle for Docker."""
+
+    def __init__(self, container: str, teardown, prompt_path_in_container: str | None):
+        self.name = container
+        self._teardown = teardown
+        self._prompt_path = prompt_path_in_container
+        self._closed = False
+
+    def exec_claude(self, argv: list[str], *, tty: bool = True) -> int:
+        full_argv = list(argv)
+        if self._prompt_path:
+            full_argv.extend(["--append-system-prompt-file", self._prompt_path])
+        cmd = ["docker", "exec"]
+        if tty:
+            cmd.append("-it")
+        cmd.extend([self.name, "claude", *full_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()

claude_bottle/backend/docker/bottle_cleanup_plan.py

@@ -0,0 +1,36 @@
+"""DockerBottleCleanupPlan — concrete subclass of BottleCleanupPlan.
+
+Holds the tuples of container and network names that
+DockerBottleBackend.cleanup will remove. The y/N preflight reads
+these via `print`; the CLI short-circuits via `empty`.
+"""
+
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass
+
+from ...log import info
+from .. import BottleCleanupPlan
+
+
+@dataclass(frozen=True)
+class DockerBottleCleanupPlan(BottleCleanupPlan):
+    """Resources DockerBottleBackend.cleanup will remove. Produced by
+    `prepare_cleanup` from a snapshot of `docker ps -a` + `docker
+    network ls`; sorted so the y/N output is stable."""
+
+    containers: tuple[str, ...]
+    networks: tuple[str, ...]
+
+    @property
+    def empty(self) -> bool:
+        return not self.containers and not self.networks
+
+    def print(self) -> None:
+        print(file=sys.stderr)
+        for name in self.containers:
+            info(f"container: {name}")
+        for name in self.networks:
+            info(f"network:   {name}")
+        print(file=sys.stderr)

claude_bottle/backend/docker/bottle_plan.py

@@ -0,0 +1,77 @@
+"""DockerBottlePlan — concrete subclass of BottlePlan.
+
+Carries the Docker-specific resolved fields produced by
+DockerBottleBackend.prepare. The launch step consumes it without
+further resolution; show_plan-style rendering is the `print` method.
+"""
+
+from __future__ import annotations
+
+import sys
+from dataclasses import dataclass
+from pathlib import Path
+
+from ...log import info
+from ...pipelock import PipelockProxyPlan
+from .. import BottlePlan
+
+
+@dataclass(frozen=True)
+class DockerBottlePlan(BottlePlan):
+    """Docker-specific resolved fields produced by
+    DockerBottleBackend.prepare. Inherits `spec` and `stage_dir` from
+    BottlePlan."""
+
+    slug: str
+    container_name: str
+    container_name_pinned: bool
+    image: str
+    derived_image: str           # "" -> no derived image
+    runtime_image: str           # image actually launched (derived or base)
+    env_file: Path
+    args_file: Path
+    prompt_file: Path
+    proxy_plan: PipelockProxyPlan
+    allowlist_summary: str
+    use_runsc: bool
+
+    def print(self, *, remote_control: bool) -> None:
+        """Render the y/N preflight summary to stderr. Pure presentation."""
+        spec = self.spec
+        manifest = spec.manifest
+        agent = manifest.agents[spec.agent_name]
+        bottle = manifest.bottle_for(spec.agent_name)
+
+        env_names = list(bottle.env.keys())
+        if spec.forward_oauth_token:
+            env_names.append("CLAUDE_CODE_OAUTH_TOKEN")
+
+        ssh_hosts = [e.Host for e in bottle.ssh]
+        prompt_first_line = agent.prompt.splitlines()[0] if agent.prompt else ""
+        runtime_label = "runsc (gVisor)" if self.use_runsc else "runc (default)"
+
+        print(file=sys.stderr)
+        info(f"agent           : {spec.agent_name}")
+        info(f"image           : {self.image}")
+        if self.derived_image:
+            info(
+                f"cwd             : {spec.user_cwd} -> /home/node/workspace "
+                f"(derived: {self.derived_image})"
+            )
+        info(f"container       : {self.container_name}")
+        info(f"stage dir       : {self.stage_dir}")
+        info("env (names only): " + (", ".join(env_names) if env_names else "(none)"))
+        info("skills          : " + (" ".join(agent.skills) if agent.skills else "(none)"))
+        info(f"docker runtime  : {runtime_label}")
+        info(f"bottle             : {agent.bottle}")
+        if ssh_hosts:
+            info(f"  ssh hosts     : {', '.join(ssh_hosts)}")
+        else:
+            info("  ssh hosts     : (none)")
+        info(f"  egress        : {self.allowlist_summary}")
+        info(
+            f"prompt          : {len(agent.prompt)} chars; "
+            f"first line: {prompt_first_line or '(empty)'}"
+        )
+        info("remote-control  : " + ("enabled" if remote_control else "disabled"))
+        print(file=sys.stderr)

claude_bottle/backend/docker/network.py

@@ -16,7 +16,7 @@ from __future__ import annotations
 
 import subprocess
 
-from .log import die, info, warn
+from ...log import die, info, warn
 
 
 def network_name_for_slug(slug: str) -> str:

claude_bottle/backend/docker/pipelock.py

@@ -0,0 +1,114 @@
+"""DockerPipelockProxy — the Docker-specific implementation of the
+sidecar's start/stop lifecycle. Inherits the platform-agnostic
+YAML-config generation from PipelockProxy."""
+
+from __future__ import annotations
+
+import os
+import subprocess
+
+from ...log import die, info, warn
+from ...pipelock import PipelockProxy, PipelockProxyPlan
+
+
+# Pipelock image, pinned by digest. The digest is the multi-arch image
+# index for ghcr.io/luckypipewrench/pipelock:2.3.0.
+PIPELOCK_IMAGE = os.environ.get(
+    "CLAUDE_BOTTLE_PIPELOCK_IMAGE",
+    "ghcr.io/luckypipewrench/pipelock@sha256:3b1a39417b98406ddc5dc2d8fcb42865ddc0c68a43d355db55f0f8cb06bc6de9",
+)
+
+# Listening port for pipelock's forward proxy.
+PIPELOCK_PORT = os.environ.get("CLAUDE_BOTTLE_PIPELOCK_PORT", "8888")
+
+
+def pipelock_container_name(slug: str) -> str:
+    return f"claude-bottle-pipelock-{slug}"
+
+
+def pipelock_proxy_url(slug: str) -> str:
+    return f"http://{pipelock_container_name(slug)}:{PIPELOCK_PORT}"
+
+
+def pipelock_proxy_host_port(slug: str) -> str:
+    return f"{pipelock_container_name(slug)}:{PIPELOCK_PORT}"
+
+
+class DockerPipelockProxy(PipelockProxy):
+    """Brings the pipelock sidecar up and down via Docker."""
+
+    def start(self, plan: PipelockProxyPlan) -> str:
+        """Boot the pipelock sidecar:
+          1. `docker create` on the internal network with the canonical
+             name and argv `run --config /etc/pipelock.yaml --listen
+             0.0.0.0:<port>`.
+          2. `docker cp` the YAML config to /etc/pipelock.yaml in the
+             writable layer (parent dir must already exist; image is
+             distroless).
+          3. Attach to the per-agent egress network.
+          4. `docker start`.
+        Returns the container name (the proxy_target passed to .stop)."""
+        name = pipelock_container_name(plan.slug)
+        if not plan.yaml_path.is_file():
+            die(
+                f"pipelock yaml not found at {plan.yaml_path}; "
+                f"PipelockProxy.prepare must run first"
+            )
+
+        info(f"starting pipelock sidecar {name} on network {plan.internal_network}")
+
+        create_args = [
+            "docker", "create",
+            "--name", name,
+            "--network", plan.internal_network,
+            PIPELOCK_IMAGE,
+            "run", "--config", "/etc/pipelock.yaml",
+            "--listen", f"0.0.0.0:{PIPELOCK_PORT}",
+        ]
+        if subprocess.run(create_args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
+            die(f"failed to create pipelock sidecar {name}")
+
+        cp_result = subprocess.run(
+            ["docker", "cp", str(plan.yaml_path), f"{name}:/etc/pipelock.yaml"],
+            capture_output=True,
+            text=True,
+        )
+        if cp_result.returncode != 0:
+            subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            die(f"failed to copy pipelock yaml into {name}: {cp_result.stderr.strip()}")
+
+        if subprocess.run(
+            ["docker", "network", "connect", plan.egress_network, name],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        ).returncode != 0:
+            subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            die(f"failed to attach pipelock sidecar {name} to egress network {plan.egress_network}")
+
+        if subprocess.run(
+            ["docker", "start", name],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        ).returncode != 0:
+            subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            die(f"failed to start pipelock sidecar {name}")
+
+        return name
+
+    def stop(self, proxy_target: str) -> None:
+        """Idempotent: missing container is success. `proxy_target` is
+        the container name returned by .start."""
+        if subprocess.run(
+            ["docker", "inspect", proxy_target],
+            stdout=subprocess.DEVNULL,
+            stderr=subprocess.DEVNULL,
+        ).returncode == 0:
+            if subprocess.run(
+                ["docker", "rm", "-f", proxy_target],
+                stdout=subprocess.DEVNULL,
+                stderr=subprocess.DEVNULL,
+            ).returncode != 0:
+                warn(
+                    f"failed to remove pipelock sidecar {proxy_target}; "
+                    f"clean up with 'docker rm -f {proxy_target}'"
+                )

claude_bottle/backend/docker/util.py

@@ -1,4 +1,6 @@
-"""Docker helpers. Build/inspect primitives shared by the CLI."""
+"""Docker host-side primitives used by DockerBottleBackend: probing
+for docker on PATH, slugifying agent names, checking image/container
+existence, and building images."""
 
 from __future__ import annotations
 
@@ -7,7 +9,18 @@ import shutil
 import subprocess
 from typing import Iterable
 
-from .log import die, info
+from ...log import die, info
+
+
+def runsc_available() -> bool:
+    """Return True if the Docker daemon has the gVisor (`runsc`) runtime
+    registered. Called once per prepare; the result lives on the plan."""
+    r = subprocess.run(
+        ["docker", "info", "--format", "{{json .Runtimes}}"],
+        capture_output=True,
+        text=True,
+    )
+    return r.returncode == 0 and "runsc" in r.stdout
 
 
 def require_docker() -> None:
@@ -19,22 +32,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
 
@@ -51,6 +48,16 @@ def container_exists(name: str) -> bool:
     return bool(result.stdout.strip())
 
 
+def docker_exec_root(container: str, argv: list[str]) -> None:
+    """Run `docker exec -u 0` in the named container, check=True. Used
+    by SSH provisioning to chown/chmod files that need root."""
+    subprocess.run(
+        ["docker", "exec", "-u", "0", container, *argv],
+        stdout=subprocess.DEVNULL,
+        check=True,
+    )
+
+
 _SLUG_RE = re.compile(r"[^a-z0-9]+")
 
 

claude_bottle/backend/util.py

@@ -0,0 +1,18 @@
+"""Cross-backend utility helpers — host-side primitives shared by
+every backend implementation. Backend-specific helpers live one level
+deeper (e.g. claude_bottle/backend/docker/util.py)."""
+
+from __future__ import annotations
+
+import os
+
+from ..log import die
+
+
+def host_skill_dir(name: str) -> str:
+    """Return the host-side path for a named skill:
+    `$HOME/.claude/skills/<name>`. Dies if HOME is unset."""
+    home = os.environ.get("HOME")
+    if not home:
+        die("HOME not set")
+    return f"{home}/.claude/skills/{name}"

claude_bottle/cli/__init__.py

@@ -1,6 +1,6 @@
 """Main CLI dispatcher.
 
-Commands: build, cleanup, edit, info, init, list, start
+Commands: cleanup, edit, info, init, list, start
 """
 
 from __future__ import annotations
@@ -9,7 +9,6 @@ import sys
 
 from ..log import Die, die
 from ._common import PROG
-from .build import cmd_build
 from .cleanup import cmd_cleanup
 from .edit import cmd_edit
 from .info import cmd_info
@@ -18,7 +17,6 @@ from .list import cmd_list
 from .start import cmd_start
 
 COMMANDS = {
-    "build": cmd_build,
     "cleanup": cmd_cleanup,
     "edit": cmd_edit,
     "info": cmd_info,
@@ -31,7 +29,6 @@ COMMANDS = {
 def usage() -> None:
     sys.stderr.write(f"usage: {PROG} <command> [args...]\n\n")
     sys.stderr.write("Commands:\n")
-    sys.stderr.write("  build     build (or rebuild) the claude-bottle Docker image\n")
     sys.stderr.write("  cleanup   stop and remove all active claude-bottle containers\n")
     sys.stderr.write("  edit      open an agent in vim for editing\n")
     sys.stderr.write("  info      print env, skills, and prompt details for a named agent\n")

claude_bottle/cli/build.py

@@ -1,19 +0,0 @@
-"""build: build (or rebuild) the claude-bottle Docker image."""
-
-from __future__ import annotations
-
-import argparse
-import os
-
-from .. import docker as docker_mod
-from ._common import PROG, REPO_DIR
-
-
-def cmd_build(argv: list[str]) -> int:
-    parser = argparse.ArgumentParser(prog=f"{PROG} build", add_help=True)
-    parser.parse_args(argv)
-
-    docker_mod.require_docker()
-    image = os.environ.get("CLAUDE_BOTTLE_IMAGE", "claude-bottle:latest")
-    docker_mod.build_image(image, REPO_DIR)
-    return 0

claude_bottle/cli/cleanup.py

@@ -1,42 +1,31 @@
-"""cleanup: stop and remove all active claude-bottle containers."""
+"""cleanup: stop and remove all orphaned claude-bottle resources
+(containers + networks) left behind by previous bottles."""
 
 from __future__ import annotations
 
-import subprocess
 import sys
 
-from .. import docker as docker_mod
+from ..backend import get_bottle_backend
 from ..log import info
 from ._common import read_tty_line
 
 
 def cmd_cleanup(_argv: list[str]) -> int:
-    docker_mod.require_docker()
-    result = subprocess.run(
-        ["docker", "ps", "--filter", "name=^claude-bottle-", "--format", "{{.Names}}"],
-        capture_output=True,
-        text=True,
-    )
-    containers = (result.stdout or "").strip()
-    if not containers:
-        info("no active claude-bottle containers")
+    backend = get_bottle_backend()
+    plan = backend.prepare_cleanup()
+
+    if plan.empty:
+        info("no claude-bottle resources to clean up")
         return 0
-    print(file=sys.stderr)
-    for name in containers.splitlines():
-        info(f"found: {name}")
-    print(file=sys.stderr)
+
+    plan.print()
     sys.stderr.write("claude-bottle: remove all of the above? [y/N] ")
     sys.stderr.flush()
     reply = read_tty_line()
     if reply not in ("y", "Y", "yes", "YES"):
         info("aborted")
         return 0
-    for name in containers.splitlines():
-        info(f"removing {name}")
-        subprocess.run(
-            ["docker", "rm", "-f", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-        )
+
+    backend.cleanup(plan)
     info("done")
     return 0

claude_bottle/cli/list.py

@@ -1,12 +1,10 @@
-"""list: list available agents or active containers."""
+"""list: list available agents or active bottles."""
 
 from __future__ import annotations
 
 import argparse
-import subprocess
 
-from .. import docker as docker_mod
-from ..log import info
+from ..backend import get_bottle_backend
 from ..manifest import Manifest
 from ._common import PROG, USER_CWD
 
@@ -22,23 +20,5 @@ def cmd_list(argv: list[str]) -> int:
             print(name)
         return 0
 
-    docker_mod.require_docker()
-    result = subprocess.run(
-        [
-            "docker", "ps",
-            "--filter", "name=^claude-bottle-",
-            "--format", "{{.Names}}\t{{.Status}}",
-        ],
-        capture_output=True,
-        text=True,
-    )
-    containers = (result.stdout or "").strip()
-    if not containers:
-        info("no active claude-bottle containers")
-        return 0
-    print()
-    for line in containers.splitlines():
-        name, _, status = line.partition("\t")
-        info(f"container: {name}  status: {status}")
-    print()
+    get_bottle_backend().list_active()
     return 0

claude_bottle/cli/start.py

@@ -7,20 +7,14 @@ 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 ..env_resolve import env_resolve
-from ..log import die, info
+from ..backend import BottleSpec, get_bottle_backend
+from ..log import 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:
@@ -33,145 +27,23 @@ def cmd_start(argv: list[str]) -> int:
 
     dry_run = args.dry_run or os.environ.get("CLAUDE_BOTTLE_DRY_RUN") == "1"
 
-    name = args.name
-    slug = docker_mod.slugify(name)
-
-    image = os.environ.get("CLAUDE_BOTTLE_IMAGE", "claude-bottle:latest")
-    default_container = f"claude-bottle-{slug}"
-    pinned_container = os.environ.get("CLAUDE_BOTTLE_CONTAINER", "")
-
-    runtime_image = image
-    derived_image = ""
-    if args.cwd:
-        derived_image = os.environ.get("CLAUDE_BOTTLE_DERIVED_IMAGE", f"claude-bottle:cwd-{slug}")
-        runtime_image = derived_image
-
-    docker_mod.require_docker()
     manifest = Manifest.resolve(USER_CWD)
-    manifest.require_agent(name)
-    agent = manifest.agents[name]
-    bottle_name = agent.bottle
-    bottle = manifest.bottle_for(name)
-
-    container = pinned_container or default_container
-    suffix = 2
-    if pinned_container:
-        if docker_mod.container_exists(container):
-            die(
-                f"container '{container}' already exists "
-                f"(pinned via CLAUDE_BOTTLE_CONTAINER). "
-                f"Remove it with 'docker rm -f {container}' or unset the override."
-            )
-    else:
-        while docker_mod.container_exists(container):
-            container = f"{default_container}-{suffix}"
-            suffix += 1
-            if suffix > 100:
-                die(
-                    f"could not find a free container name after "
-                    f"{default_container}-99; clean up old containers with "
-                    f"'docker rm -f <name>'"
-                )
-
-    # --- Plan resolution (host-only, no container yet) ---
-    env_names = list(bottle.env.keys())
-
-    # CLAUDE_BOTTLE_OAUTH_TOKEN → CLAUDE_CODE_OAUTH_TOKEN forwarding.
-    # Host-side token is always forwarded so every container can authenticate.
-    forward_oauth_token = bool(os.environ.get("CLAUDE_BOTTLE_OAUTH_TOKEN"))
-    display_env_names = list(env_names)
-    if forward_oauth_token:
-        display_env_names.append("CLAUDE_CODE_OAUTH_TOKEN")
-
-    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)
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=args.name,
+        copy_cwd=args.cwd,
+        user_cwd=USER_CWD,
+        forward_oauth_token=bool(os.environ.get("CLAUDE_BOTTLE_OAUTH_TOKEN")),
+    )
 
     stage_dir = Path(tempfile.mkdtemp(prefix="claude-bottle-stage."))
-    env_file = stage_dir / "agent.env"
-    args_file = stage_dir / "docker-args"
-    prompt_file = stage_dir / "prompt.txt"
-    pipelock_yaml_filename = "pipelock.yaml"
-    pipelock_yaml = stage_dir / pipelock_yaml_filename
-    env_file.write_text("")
-    env_file.chmod(0o600)
-    args_file.write_text("")
-    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)
-
-        env_resolve(manifest, name, env_file, args_file)
-
-        prompt_content = agent.prompt
-        prompt_file.write_text(prompt_content)
-        prompt_first_line = prompt_content.splitlines()[0] if prompt_content else ""
-
-        # --- Plan + confirm ---
-        print(file=sys.stderr)
-        info(f"agent           : {name}")
-        info(f"image           : {image}")
-        if derived_image:
-            info(f"cwd             : {USER_CWD} -> /home/node/workspace (derived: {derived_image})")
-        info(f"container       : {container}")
-        info(f"stage dir       : {stage_dir}")
-        info(
-            "env (names only): "
-            + (", ".join(display_env_names) if display_env_names else "(none)")
-        )
-        info("skills          : " + (" ".join(agent.skills) if agent.skills else "(none)"))
-        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}")
-        else:
-            info("  ssh hosts     : (none)")
-        info(f"  egress        : {allowlist_summary}")
-        info(
-            f"prompt          : {len(prompt_content)} chars; "
-            f"first line: {prompt_first_line or '(empty)'}"
-        )
-        info("remote-control  : " + ("enabled" if args.remote_control else "disabled"))
-        print(file=sys.stderr)
+        backend = get_bottle_backend()
+        plan = backend.prepare(spec, stage_dir=stage_dir)
+        plan.print(remote_control=args.remote_control)
 
         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 +51,18 @@ 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,
-        )
-
-        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"
-                )
-            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,
+        with backend.launch(plan) as bottle:
+            info(
+                "attaching interactive claude session "
+                "(Ctrl-D or 'exit' to leave; container will be removed)"
             )
-            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
+            claude_args = ["--dangerously-skip-permissions"]
+            if args.remote_control:
+                claude_args.append("--remote-control")
+            bottle.exec_claude(claude_args, tty=True)
+            info(f"session ended; container {bottle.name} will be removed")
+            return 0
     finally:
-        cleanup_all()
+        shutil.rmtree(stage_dir, ignore_errors=True)

claude_bottle/env.py

@@ -1,28 +1,27 @@
-"""Env resolver. Walks the env entries for one agent and produces:
+"""Env resolver. Walks the env entries for one agent and produces a
+backend-neutral ResolvedEnv describing how the bottle should receive
+each variable:
 
-  1. The list of `docker run` arg fragments needed to forward each var.
-     Both `secret` and `interpolated` entries become `-e NAME` (no
-     `=value`) so Docker inherits the value from this process env
-     without rendering it on argv or persisting it to disk.
-     Only `literal` entries are written to a host-disk env-file.
-  2. The export side-effect of populating this process's env with
-     secret values prompted from the user, and with interpolated
-     values copied from the matching host var, so `-e NAME` actually
-     has something to inherit.
+  - `forwarded` — names whose values have been placed into this
+    process's env (from a tty prompt for `secret`, from the matching
+    host var for `interpolated`). The backend is expected to pass
+    these to the bottle by-name so the value never appears on argv,
+    in a file, or in a log line.
+  - `literals` — name→value pairs that the manifest carries verbatim.
+    The backend serializes these however its launcher accepts env
+    (an env-file, an API payload, etc.).
 
 Each env entry is a string. Mode is selected by sentinel prefix:
-  "?"           → secret (prompt at runtime). Bare "?" uses default prompt;
+  "?"           -> secret (prompt at runtime). Bare "?" uses default prompt;
                   "?<message>" uses <message> as the prompt body.
-  "${HOST_VAR}" → interpolated from $HOST_VAR in the host process env
-  any other str → literal (the string is the value verbatim)
+  "${HOST_VAR}" -> interpolated from $HOST_VAR in the host process env
+  any other str -> literal (the string is the value verbatim)
 
 Critical rules:
   - NEVER echo, log, or interpolate the value of a secret or
     interpolated env var. Both are treated as potentially sensitive:
     nothing about their value (other than presence) ever lands on
     disk, in a log line, or on argv.
-  - The env-file written for literals lives under mktemp -d with mode
-    600, removed by the caller's cleanup.
   - Errors mention only the variable NAME, never any portion of the value.
 """
 
@@ -32,7 +31,7 @@ import getpass
 import os
 import re
 import sys
-from pathlib import Path
+from dataclasses import dataclass, field
 
 from .log import die
 from .manifest import Manifest
@@ -40,6 +39,18 @@ from .manifest import Manifest
 _INTERPOLATED_RE = re.compile(r"^\$\{([A-Za-z_][A-Za-z0-9_]*)\}$")
 
 
+@dataclass(frozen=True)
+class ResolvedEnv:
+    """Backend-neutral env resolution result.
+
+    `forwarded` names have already been exported into os.environ by
+    resolve_env; the backend forwards by-name. `literals` carry their
+    values verbatim and are serialized by the backend."""
+
+    forwarded: list[str] = field(default_factory=list)
+    literals: dict[str, str] = field(default_factory=dict)
+
+
 def env_entry_kind(raw: str) -> str:
     """Returns 'secret', 'interpolated', or 'literal'."""
     if raw.startswith("?"):
@@ -97,17 +108,14 @@ def _read_secret_silent(name: str, prompt_body: str) -> str:
     return value
 
 
-def env_resolve(
-    manifest: Manifest,
-    agent: str,
-    env_file: Path,
-    out_args: Path,
-) -> None:
+def resolve_env(manifest: Manifest, agent: str) -> ResolvedEnv:
     """Iterate the agent's env entries:
-      - secret:       always prompt; export into this process; append `-e NAME` to out_args
-      - interpolated: copy host value; export under target name; append `-e NAME`
-      - literal:      append `NAME=VALUE` to env_file
+      - secret:       always prompt; export into this process; mark forwarded
+      - interpolated: copy host value; export under target name; mark forwarded
+      - literal:      include in the literals map verbatim
     """
+    forwarded: list[str] = []
+    literals: dict[str, str] = {}
     bottle = manifest.bottle_for(agent)
     for name, raw in bottle.env.items():
         if not name:
@@ -117,8 +125,7 @@ def env_resolve(
             prompt_body = env_entry_secret_prompt(raw)
             value = _read_secret_silent(name, prompt_body)
             os.environ[name] = value
-            with out_args.open("a") as f:
-                f.write(f"-e\n{name}\n")
+            forwarded.append(name)
         elif kind == "interpolated":
             host_var = env_entry_interpolated_from(raw)
             host_value = os.environ.get(host_var, "")
@@ -128,13 +135,7 @@ def env_resolve(
                     f"but ${host_var} is unset or empty in the host environment."
                 )
             os.environ[name] = host_value
-            with out_args.open("a") as f:
-                f.write(f"-e\n{name}\n")
+            forwarded.append(name)
         else:  # literal
-            if "\n" in raw:
-                die(
-                    f"env entry {name} (literal) contains a newline; "
-                    f"docker --env-file cannot represent multi-line values."
-                )
-            with env_file.open("a") as f:
-                f.write(f"{name}={raw}\n")
+            literals[name] = raw
+    return ResolvedEnv(forwarded=forwarded, literals=literals)

claude_bottle/manifest.py

@@ -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 by the "
+                f"backend; remove the 'runtime' field from the bottle "
+                f"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)

claude_bottle/pipelock.py

@@ -12,23 +12,12 @@ Image pin: ghcr.io/luckypipewrench/pipelock@sha256:<digest> for tag 2.3.0.
 
 from __future__ import annotations
 
-import os
-import re
-import subprocess
+from abc import ABC, abstractmethod
+from dataclasses import dataclass
 from pathlib import Path
 
-from .log import die, info, warn
-from .manifest import Manifest
-
-# Pipelock image, pinned by digest. The digest is the multi-arch image
-# index for ghcr.io/luckypipewrench/pipelock:2.3.0.
-PIPELOCK_IMAGE = os.environ.get(
-    "CLAUDE_BOTTLE_PIPELOCK_IMAGE",
-    "ghcr.io/luckypipewrench/pipelock@sha256:3b1a39417b98406ddc5dc2d8fcb42865ddc0c68a43d355db55f0f8cb06bc6de9",
-)
-
-# Listening port for pipelock's forward proxy.
-PIPELOCK_PORT = os.environ.get("CLAUDE_BOTTLE_PIPELOCK_PORT", "8888")
+from .manifest import Bottle
+from .util import is_ipv4_literal
 
 # Baked-in default allowlist for hosts Claude Code itself needs.
 DEFAULT_ALLOWLIST: tuple[str, ...] = (
@@ -42,69 +31,45 @@ DEFAULT_ALLOWLIST: tuple[str, ...] = (
 )
 
 
-def pipelock_container_name(slug: str) -> str:
-    return f"claude-bottle-pipelock-{slug}"
-
-
-def pipelock_proxy_url(slug: str) -> str:
-    return f"http://{pipelock_container_name(slug)}:{PIPELOCK_PORT}"
-
-
-def pipelock_proxy_host_port(slug: str) -> str:
-    return f"{pipelock_container_name(slug)}:{PIPELOCK_PORT}"
-
-
 # --- Allowlist resolution --------------------------------------------------
 
 
-def pipelock_bottle_allowlist(manifest: Manifest, bottle_name: str) -> list[str]:
-    """Hostnames in bottles[<bottle_name>].egress.allowlist."""
-    return list(manifest.bottles[bottle_name].egress.allowlist)
+def pipelock_bottle_allowlist(bottle: Bottle) -> list[str]:
+    """Hostnames in bottle.egress.allowlist."""
+    return list(bottle.egress.allowlist)
 
 
-def pipelock_bottle_ssh_hostnames(manifest: Manifest, bottle_name: str) -> list[str]:
-    return [e.Hostname for e in manifest.bottles[bottle_name].ssh if e.Hostname]
+def pipelock_bottle_ssh_hostnames(bottle: Bottle) -> list[str]:
+    return [e.Hostname for e in bottle.ssh if e.Hostname]
 
 
-_IPV4_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$")
+def pipelock_bottle_ssh_trusted_domains(bottle: Bottle) -> list[str]:
+    return [h for h in pipelock_bottle_ssh_hostnames(bottle) if not is_ipv4_literal(h)]
 
 
-def is_ipv4_literal(s: str) -> bool:
-    """Pipelock's SSRF check fires on resolved IP, so an IP-literal
-    Hostname goes to ssrf.ip_allowlist while a hostname goes to
-    trusted_domains."""
-    if not s:
-        return False
-    return bool(_IPV4_RE.match(s))
+def pipelock_bottle_ssh_ip_cidrs(bottle: Bottle) -> list[str]:
+    return [f"{h}/32" for h in pipelock_bottle_ssh_hostnames(bottle) if is_ipv4_literal(h)]
 
 
-def pipelock_bottle_ssh_trusted_domains(manifest: Manifest, bottle_name: str) -> list[str]:
-    return [h for h in pipelock_bottle_ssh_hostnames(manifest, bottle_name) if not is_ipv4_literal(h)]
-
-
-def pipelock_bottle_ssh_ip_cidrs(manifest: Manifest, bottle_name: str) -> list[str]:
-    return [f"{h}/32" for h in pipelock_bottle_ssh_hostnames(manifest, bottle_name) if is_ipv4_literal(h)]
-
-
-def pipelock_effective_allowlist(manifest: Manifest, bottle_name: str) -> list[str]:
+def pipelock_effective_allowlist(bottle: Bottle) -> list[str]:
     """Deduplicated union of: baked-in defaults, bottle.egress.allowlist,
     bottle.ssh[].Hostname. Sorted for stability."""
     seen: dict[str, None] = {}
     for h in DEFAULT_ALLOWLIST:
         seen.setdefault(h, None)
-    for h in pipelock_bottle_allowlist(manifest, bottle_name):
+    for h in pipelock_bottle_allowlist(bottle):
         if h:
             seen.setdefault(h, None)
-    for h in pipelock_bottle_ssh_hostnames(manifest, bottle_name):
+    for h in pipelock_bottle_ssh_hostnames(bottle):
         if h:
             seen.setdefault(h, None)
     return sorted(seen.keys())
 
 
-def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str:
+def pipelock_allowlist_summary(bottle: Bottle) -> str:
     """One-line summary for the y/N preflight display:
     "<N> hosts allowed (host1, host2, host3, +M more)"."""
-    hosts = pipelock_effective_allowlist(manifest, bottle_name)
+    hosts = pipelock_effective_allowlist(bottle)
     count = len(hosts)
     if count == 0:
         return "0 hosts allowed (none)"
@@ -119,129 +84,98 @@ def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str:
     return f"{count} hosts allowed ({joined})"
 
 
-# --- YAML generation -------------------------------------------------------
+
+# --- Proxy class -----------------------------------------------------------
 
 
-def pipelock_write_yaml(manifest: Manifest, bottle_name: str, out_path: Path) -> None:
-    """Write a pipelock YAML config (mode 600) carrying:
-      - the effective allowlist (hostnames),
-      - a fixed listen port,
-      - strict mode + forward_proxy.enabled + DLP defaults + scan_env.
+@dataclass(frozen=True)
+class PipelockProxyPlan:
+    """Output of PipelockProxy.prepare; consumed by .start when the
+    sidecar needs to be brought up.
 
-    Deliberately contains no env values, no secrets, no per-agent
-    customization beyond the hostname list."""
-    allowlist = pipelock_effective_allowlist(manifest, bottle_name)
-    trusted = pipelock_bottle_ssh_trusted_domains(manifest, bottle_name)
-    ip_cidrs = pipelock_bottle_ssh_ip_cidrs(manifest, bottle_name)
+    yaml_path + slug are filled in at prepare time. internal_network
+    and egress_network default to empty and are populated by the
+    backend's launch step (via dataclasses.replace) once those networks
+    have actually been created."""
 
-    lines: list[str] = []
-    lines.append("version: 1")
-    lines.append("mode: strict")
-    lines.append("enforce: true")
-    lines.append("")
-    lines.append("# Hostnames the agent is allowed to reach. Effective list is")
-    lines.append("# claude-bottle defaults UNION bottle.egress.allowlist (sorted, deduped).")
-    lines.append("api_allowlist:")
-    for h in allowlist:
-        lines.append(f'  - "{h}"')
-    lines.append("")
-    lines.append("forward_proxy:")
-    lines.append("  enabled: true")
-    lines.append("")
-    if trusted:
-        lines.append("trusted_domains:")
-        for td in trusted:
-            lines.append(f'  - "{td}"')
+    yaml_path: Path
+    slug: str
+    internal_network: str = ""
+    egress_network: str = ""
+
+
+class PipelockProxy(ABC):
+    """The pipelock egress proxy. Encapsulates the YAML-config
+    generation; the sidecar's start/stop lifecycle is backend-specific
+    and lives on concrete subclasses."""
+
+    def prepare(
+        self, bottle: Bottle, slug: str, yaml_path: Path
+    ) -> PipelockProxyPlan:
+        """Write the pipelock yaml config (mode 600) to `yaml_path`
+        and return the plan for `.start`.
+
+        `slug` is the agent-derived identifier (lowercased,
+        hyphen-normalized) used as the suffix in every per-agent
+        resource name — the agent container, the pipelock container
+        (`claude-bottle-pipelock-<slug>`), the internal/egress
+        networks. It's stored on the returned plan so the backend's
+        start step can derive the sidecar's container name."""
+        self._build_pipelock_yaml(bottle, yaml_path)
+        return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
+
+    def _build_pipelock_yaml(self, bottle: Bottle, yaml_path: Path):
+        """Write the pipelock yaml config (mode 600) to `yaml_path`
+        for the sidecar to consume when it boots. Carries the
+        effective allowlist (bottle.egress.allowlist UNION
+        claude-bottle defaults UNION ssh hostnames), a fixed listen
+        port, strict mode + forward_proxy + DLP defaults + scan_env.
+        Deliberately contains no env values, no secrets, no per-agent
+        customization beyond the hostname list."""
+        allowlist = pipelock_effective_allowlist(bottle)
+        trusted = pipelock_bottle_ssh_trusted_domains(bottle)
+        ip_cidrs = pipelock_bottle_ssh_ip_cidrs(bottle)
+
+        lines: list[str] = []
+        lines.append("version: 1")
+        lines.append("mode: strict")
+        lines.append("enforce: true")
         lines.append("")
-    if ip_cidrs:
-        lines.append("ssrf:")
-        lines.append("  ip_allowlist:")
-        for cidr in ip_cidrs:
-            lines.append(f'    - "{cidr}"')
+        lines.append("# Hostnames the agent is allowed to reach. Effective list is")
+        lines.append("# claude-bottle defaults UNION bottle.egress.allowlist (sorted, deduped).")
+        lines.append("api_allowlist:")
+        for h in allowlist:
+            lines.append(f'  - "{h}"')
         lines.append("")
-    lines.append("dlp:")
-    lines.append("  include_defaults: true")
-    lines.append("  scan_env: true")
+        lines.append("forward_proxy:")
+        lines.append("  enabled: true")
+        lines.append("")
+        if trusted:
+            lines.append("trusted_domains:")
+            for td in trusted:
+                lines.append(f'  - "{td}"')
+            lines.append("")
+        if ip_cidrs:
+            lines.append("ssrf:")
+            lines.append("  ip_allowlist:")
+            for cidr in ip_cidrs:
+                lines.append(f'    - "{cidr}"')
+            lines.append("")
+        lines.append("dlp:")
+        lines.append("  include_defaults: true")
+        lines.append("  scan_env: true")
 
-    out_path.write_text("\n".join(lines) + "\n")
-    out_path.chmod(0o600)
+        yaml_path.write_text("\n".join(lines) + "\n")
+        yaml_path.chmod(0o600)
 
+    @abstractmethod
+    def start(self, plan: PipelockProxyPlan) -> str:
+        """Bring up the pipelock sidecar according to `plan`. Returns
+        the proxy_target string identifying the running instance — the
+        same value to pass to `.stop`. Backend-specific."""
 
-# --- Sidecar lifecycle -----------------------------------------------------
-
-
-def pipelock_start(
-    slug: str,
-    internal_network: str,
-    egress_network: str,
-    yaml_dir: Path,
-    yaml_filename: str,
-) -> str:
-    """Boot the pipelock sidecar:
-      1. `docker create` on the internal network with the canonical name
-         and argv `run --config /etc/pipelock.yaml --listen 0.0.0.0:<port>`.
-      2. `docker cp` the YAML config to /etc/pipelock.yaml in the
-         writable layer (parent dir must already exist; image is distroless).
-      3. Attach to the per-agent egress network.
-      4. `docker start`.
-    Returns the container name."""
-    name = pipelock_container_name(slug)
-    host_yaml = yaml_dir / yaml_filename
-    if not host_yaml.is_file():
-        die(f"pipelock yaml not found at {host_yaml}; pipelock_write_yaml must run first")
-
-    info(f"starting pipelock sidecar {name} on network {internal_network}")
-
-    create_args = [
-        "docker", "create",
-        "--name", name,
-        "--network", internal_network,
-        PIPELOCK_IMAGE,
-        "run", "--config", "/etc/pipelock.yaml",
-        "--listen", f"0.0.0.0:{PIPELOCK_PORT}",
-    ]
-    if subprocess.run(create_args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
-        die(f"failed to create pipelock sidecar {name}")
-
-    cp_result = subprocess.run(
-        ["docker", "cp", str(host_yaml), f"{name}:/etc/pipelock.yaml"],
-        capture_output=True,
-        text=True,
-    )
-    if cp_result.returncode != 0:
-        subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-        die(f"failed to copy pipelock yaml into {name}: {cp_result.stderr.strip()}")
-
-    if subprocess.run(
-        ["docker", "network", "connect", egress_network, name],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    ).returncode != 0:
-        subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-        die(f"failed to attach pipelock sidecar {name} to egress network {egress_network}")
-
-    if subprocess.run(
-        ["docker", "start", name],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    ).returncode != 0:
-        subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
-        die(f"failed to start pipelock sidecar {name}")
-
-    return name
-
-
-def pipelock_stop(slug: str) -> None:
-    """Idempotent: missing container is success."""
-    name = pipelock_container_name(slug)
-    if subprocess.run(
-        ["docker", "inspect", name],
-        stdout=subprocess.DEVNULL,
-        stderr=subprocess.DEVNULL,
-    ).returncode == 0:
-        if subprocess.run(
-            ["docker", "rm", "-f", name],
-            stdout=subprocess.DEVNULL,
-            stderr=subprocess.DEVNULL,
-        ).returncode != 0:
-            warn(f"failed to remove pipelock sidecar {name}; clean up with 'docker rm -f {name}'")
+    @abstractmethod
+    def stop(self, proxy_target: str) -> None:
+        """Tear down the pipelock sidecar identified by `proxy_target`
+        (the value `.start` returned). Idempotent: a missing target is
+        success. Backend-specific."""

claude_bottle/skills.py

@@ -1,76 +0,0 @@
-"""Skill copier: host's ~/.claude/skills/<name>/ -> container's
-~/.claude/skills/<name>/, preserving directory structure."""
-
-from __future__ import annotations
-
-import os
-import subprocess
-
-from .log import die, info
-
-CONTAINER_HOME = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
-CONTAINER_SKILLS_DIR = os.environ.get(
-    "CLAUDE_BOTTLE_CONTAINER_SKILLS_DIR", f"{CONTAINER_HOME}/.claude/skills"
-)
-
-
-def host_skill_dir(name: str) -> str:
-    home = os.environ.get("HOME")
-    if not home:
-        die("HOME not set")
-    return f"{home}/.claude/skills/{name}"
-
-
-def host_skill_exists(name: str) -> bool:
-    return os.path.isdir(host_skill_dir(name))
-
-
-def require_host_skill(name: str) -> None:
-    if not host_skill_exists(name):
-        die(
-            f"skill '{name}' not found on host at {host_skill_dir(name)}. "
-            f"Create it under ~/.claude/skills/, then re-run."
-        )
-
-
-def skills_validate_all(names: list[str]) -> None:
-    """Use BEFORE the y/N so the user does not get asked about a plan
-    that's already known to fail."""
-    for n in names:
-        require_host_skill(n)
-
-
-def skills_copy_into(container: str, names: list[str]) -> None:
-    """For each named skill, ensure the parent dir exists, wipe any
-    prior copy, then `docker cp <host>/. <container>:<dst>/` so the
-    contents are copied into a freshly-created destination dir."""
-    if not names:
-        return
-
-    subprocess.run(
-        ["docker", "exec", container, "mkdir", "-p", CONTAINER_SKILLS_DIR],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-
-    for n in names:
-        src = host_skill_dir(n)
-        if not os.path.isdir(src):
-            die(f"skill '{n}' disappeared from host between validation and copy at {src}.")
-        dst = f"{CONTAINER_SKILLS_DIR}/{n}"
-        info(f"copying skill {n} into {container}:{dst}")
-        subprocess.run(
-            ["docker", "exec", container, "rm", "-rf", dst],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        subprocess.run(
-            ["docker", "exec", container, "mkdir", "-p", dst],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        subprocess.run(
-            ["docker", "cp", f"{src}/.", f"{container}:{dst}/"],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )

claude_bottle/ssh.py

@@ -1,204 +0,0 @@
-"""SSH helpers. Validates ssh entries from claude-bottle.json, then sets
-up SSH inside the container via a root-owned ssh-agent so the `node`
-user can use the keys for SSH but cannot read the key bytes.
-
-Why an in-container agent (not bind-mounted from host): Docker Desktop
-on macOS does not forward Unix-domain socket connect() across the VM
-boundary — connect() returns ENOTSUP. Running ssh-agent inside the
-container sidesteps that entirely.
-
-Isolation:
-  - Keys live at /root/.claude-bottle-keys/ (mode 700, root-owned).
-    /root is mode 700 in node:22-slim, so node (uid 1000) can't even
-    traverse in.
-  - ssh-agent runs as root, listening on /run/claude-bottle-agent.sock.
-    Each key is loaded with ssh-add, then deleted; the bytes now live
-    only in the agent process's memory.
-  - ssh-agent's SO_PEERCRED-based UID match rejects every connection
-    whose peer euid is neither 0 nor the agent's. To bridge that, a
-    root-owned socat forwarder listens on
-    /run/claude-bottle-agent-public.sock (mode 666) and proxies bytes
-    to the real agent socket.
-  - node can't ptrace root-owned agent or socat, so /proc/<pid>/mem is
-    off-limits and key bytes never leave root-owned memory.
-  - ~/.ssh/config in node's home points each Host at the public socket
-    via IdentityAgent.
-
-Limitation: keys must be passphrase-less. ssh-add prompts on /dev/tty
-for passphrases, but our docker exec has no TTY.
-
-Each ssh entry has keys: Host, IdentityFile, Hostname, User, Port
-(required); KnownHostKey (optional).
-"""
-
-from __future__ import annotations
-
-import os
-import subprocess
-from pathlib import Path
-from typing import Sequence
-
-from .log import die, info
-from .manifest import SshEntry
-
-
-def ssh_validate_entries(entries: Sequence[SshEntry]) -> None:
-    """The IdentityFile must exist on the host (after expanding leading ~).
-    Host and IdentityFile shape are already enforced by Manifest validation."""
-    for entry in entries:
-        key = _expand_tilde(entry.IdentityFile)
-        if not os.path.isfile(key):
-            die(f"ssh key file not found for host '{entry.Host}': {key}")
-
-
-def ssh_setup(
-    container: str,
-    stage_dir: Path,
-    proxy_host_port: str,
-    entries: Sequence[SshEntry],
-) -> None:
-    """Set up SSH in the container so node can authenticate using each
-    entry's key without the key file being readable by node."""
-    container_home = os.environ.get("CLAUDE_BOTTLE_CONTAINER_HOME", "/home/node")
-    container_ssh = f"{container_home}/.ssh"
-    agent_socket = "/run/claude-bottle-agent.sock"
-    public_socket = "/run/claude-bottle-agent-public.sock"
-    keys_dir = "/root/.claude-bottle-keys"
-
-    # ~/.ssh for node (700, owned by node).
-    _docker_exec_root(container, ["mkdir", "-p", container_ssh])
-    _docker_exec_root(container, ["chown", "node:node", container_ssh])
-    _docker_exec_root(container, ["chmod", "700", container_ssh])
-
-    # /root/.claude-bottle-keys for root (700, root-owned).
-    _docker_exec_root(container, ["mkdir", "-p", keys_dir])
-    _docker_exec_root(container, ["chown", "root:root", keys_dir])
-    _docker_exec_root(container, ["chmod", "700", keys_dir])
-
-    config_file = stage_dir / "ssh_config"
-    known_hosts_file = stage_dir / "ssh_known_hosts"
-    config_file.write_text("")
-    config_file.chmod(0o600)
-    known_hosts_file.write_text("")
-    known_hosts_file.chmod(0o600)
-
-    proxy_host, _, proxy_port = proxy_host_port.partition(":")
-
-    container_key_paths: list[str] = []
-    for entry in entries:
-        name = entry.Host
-        key = _expand_tilde(entry.IdentityFile)
-        hostname = entry.Hostname
-        user = entry.User
-        port = entry.Port
-        known_host_key = entry.KnownHostKey
-
-        key_basename = os.path.basename(key)
-        container_key_path = f"{keys_dir}/{key_basename}"
-
-        info(f"copying ssh key for '{name}' -> {container} (root-only staging)")
-        subprocess.run(
-            ["docker", "cp", key, f"{container}:{container_key_path}"],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        _docker_exec_root(container, ["chown", "root:root", container_key_path])
-        _docker_exec_root(container, ["chmod", "600", container_key_path])
-
-        container_key_paths.append(container_key_path)
-
-        # ProxyCommand tunnels SSH through pipelock via HTTP CONNECT.
-        # %h / %p expand to this block's HostName / Port. socat's
-        # PROXY: mode does CONNECT host:port to the proxy.
-        block = (
-            f"Host {name}\n"
-            f"  HostName {hostname}\n"
-            f"  User {user}\n"
-            f"  Port {port}\n"
-            f"  IdentityAgent {public_socket}\n"
-            f"  ProxyCommand socat - PROXY:{proxy_host}:%h:%p,proxyport={proxy_port}\n"
-            f"\n"
-        )
-        with config_file.open("a") as f:
-            f.write(block)
-
-        if known_host_key:
-            entries_to_write: list[str] = []
-            if port == "22":
-                entries_to_write.append(f"{name} {known_host_key}\n")
-                if hostname != name:
-                    entries_to_write.append(f"{hostname} {known_host_key}\n")
-            else:
-                entries_to_write.append(f"[{name}]:{port} {known_host_key}\n")
-                if hostname != name:
-                    entries_to_write.append(f"[{hostname}]:{port} {known_host_key}\n")
-            with known_hosts_file.open("a") as f:
-                for e in entries_to_write:
-                    f.write(e)
-
-    # Boot the agent, load each key, delete the key files, then start
-    # the root-owned socat forwarder. One docker exec so the whole
-    # sequence is atomic.
-    info(f"starting in-container ssh-agent at {agent_socket} (forwarded via {public_socket})")
-    setup_lines = [
-        "set -eu",
-        f"ssh-agent -a {agent_socket} >/dev/null",
-    ]
-    for kp in container_key_paths:
-        setup_lines.append(f"SSH_AUTH_SOCK={agent_socket} ssh-add {kp}")
-        setup_lines.append(f"rm -f {kp}")
-    setup_lines.append(f"rmdir {keys_dir} 2>/dev/null || true")
-    # Forwarder: socat (uid 0) connects to the agent on node's behalf.
-    setup_lines.append(
-        f"nohup socat UNIX-LISTEN:{public_socket},fork,reuseaddr,mode=666 "
-        f"UNIX-CONNECT:{agent_socket} </dev/null >/dev/null 2>&1 &"
-    )
-    # Wait briefly for the forwarder to bind.
-    setup_lines.extend([
-        "i=0",
-        "while [ $i -lt 20 ]; do",
-        f"  [ -S {public_socket} ] && break",
-        "  i=$((i + 1))",
-        "  sleep 0.1",
-        "done",
-        f"[ -S {public_socket} ] || {{ echo 'claude-bottle: socat forwarder failed to bind {public_socket}' >&2; exit 1; }}",
-    ])
-    setup_script = "\n".join(setup_lines) + "\n"
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, "sh", "-c", setup_script],
-        check=True,
-    )
-
-    info(f"writing {container_ssh}/config")
-    subprocess.run(
-        ["docker", "cp", str(config_file), f"{container}:{container_ssh}/config"],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-    _docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/config"])
-    _docker_exec_root(container, ["chmod", "600", f"{container_ssh}/config"])
-
-    if known_hosts_file.stat().st_size > 0:
-        info(f"writing {container_ssh}/known_hosts")
-        subprocess.run(
-            ["docker", "cp", str(known_hosts_file), f"{container}:{container_ssh}/known_hosts"],
-            stdout=subprocess.DEVNULL,
-            check=True,
-        )
-        _docker_exec_root(container, ["chown", "node:node", f"{container_ssh}/known_hosts"])
-        _docker_exec_root(container, ["chmod", "600", f"{container_ssh}/known_hosts"])
-
-
-def _docker_exec_root(container: str, argv: list[str]) -> None:
-    subprocess.run(
-        ["docker", "exec", "-u", "0", container, *argv],
-        stdout=subprocess.DEVNULL,
-        check=True,
-    )
-
-
-def _expand_tilde(path: str) -> str:
-    if path.startswith("~"):
-        home = os.environ.get("HOME", "")
-        return home + path[1:]
-    return path

claude_bottle/util.py

@@ -0,0 +1,31 @@
+"""Cross-cutting utility helpers used by multiple modules.
+
+Top-level (i.e. backend-agnostic) — backend-specific helpers live one
+level deeper, under their backend package."""
+
+from __future__ import annotations
+
+import os
+import re
+
+
+def expand_tilde(path: str) -> str:
+    """Expand a leading '~' to $HOME. Leaves paths without a leading
+    tilde unchanged. Falls back to the empty string if $HOME is unset
+    (callers should already have checked HOME if they care)."""
+    if path.startswith("~"):
+        home = os.environ.get("HOME", "")
+        return home + path[1:]
+    return path
+
+
+_IPV4_RE = re.compile(r"^[0-9]+\.[0-9]+\.[0-9]+\.[0-9]+$")
+
+
+def is_ipv4_literal(s: str) -> bool:
+    """True iff `s` looks like a dotted-quad IPv4 literal. Does not
+    validate octet ranges; consumers that care about that should run
+    a stricter check. Empty input returns False."""
+    if not s:
+        return False
+    return bool(_IPV4_RE.match(s))

cli.py

@@ -1,19 +1,6 @@
 #!/usr/bin/env python3
-"""cli.py — manage claude-bottle containers.
-
-usage: cli.py <command> [args...]
-
-Commands:
-  build     build (or rebuild) the claude-bottle Docker image.
-  cleanup   stop and remove all active claude-bottle containers.
-  edit      open an agent in vim for editing.
-  info      print env, skills, and prompt details for a named agent.
-  init      interactively create a new agent and add it to claude-bottle.json.
-  list      list available agents or active containers.
-  start     boot a sandboxed container for a named agent and attach an
-            interactive claude-code session. The container is torn down
-            when the session ends.
-"""
+"""cli.py — entry point for the claude-bottle CLI. Run with --help (or
+no args) for the command list."""
 
 from __future__ import annotations
 

docs/prds/0003-bottle-backend-abstraction.md

@@ -0,0 +1,300 @@
+# PRD 0003: Bottle Backend abstraction
+
+- **Status:** Draft
+- **Author:** didericis
+- **Created:** 2026-05-10
+
+## Summary
+
+Introduce a per-backend abstraction that owns the end-to-end lifecycle
+of a "bottle" (a running, isolated environment with claude inside).
+The first and only implementation lands as `DockerBottleBackend`. No
+second backend ships in this PRD.
+
+## Problem
+
+Today, "how to launch a bottle" is spread across roughly six modules
+(`claude_bottle/cli/start.py`, `pipelock.py`, `network.py`, `ssh.py`,
+`skills.py`, `docker.py`), each shelling out to `docker` directly via
+`subprocess.run(["docker", ...])`. That coupling means:
+
+- Adding a second backend (Apple's `container`, fly.io, a remote SSH
+  host, etc.) requires editing every one of those call sites. The
+  research note `docs/research/apple-container-backend.md` already
+  flags this as a prerequisite for that work.
+- The pipelock sidecar topology — two networks, multi-attach, sidecar
+  lifecycle — is a Docker implementation detail that has leaked into
+  the top-level CLI orchestration. It reads as a core concept of the
+  project, but a fly.io bottle would not need any of it.
+- The manifest carries a Docker-specific `runtime: "runsc"` field
+  (`bottles[].runtime`). Anyone setting it has to know about gVisor,
+  whether Docker has it registered, and what to do on macOS where it
+  isn't available natively. The field has one valid non-default value
+  and exists only because the current code can't decide on its own.
+
+The shape that fits the project's actual goals (isolated agent runs
+across multiple backends) is "one backend per platform," not "one
+container-runtime SDK with N drivers." A previous draft of this PRD
+considered a low-level runtime-primitive protocol (`run`, `exec`,
+`cp`, `network_connect`, ...) and rejected it as the wrong layer —
+it would have forced fly.io to pretend it's Docker.
+
+## Goals / Success Criteria
+
+The feature works when all of the following are observable:
+
+- `cli.py start` works identically for an existing manifest with no
+  user-visible changes other than (a) a startup log line naming the
+  Docker runtime in use, and (b) `bottles[].runtime` no longer being a
+  valid manifest field.
+- On a Linux host with gVisor registered, the agent container runs
+  under `runsc` without anything in the manifest requesting it.
+- On a host without gVisor (including macOS), the agent container runs
+  under the default `runc` runtime; nothing fails, no warning is
+  printed beyond the runtime-name log line.
+- The existing test suite passes with no behavior changes other than
+  the manifest-schema removal of `runtime`.
+
+The feature is **done** when all of the following ship:
+
+- A new `claude_bottle/backend/` package exists with abstract base
+  classes (`BottleBackend`, `BottlePlan`, `BottleCleanupPlan`,
+  `Bottle`) plus a `claude_bottle/backend/docker/` subpackage
+  containing the `DockerBottleBackend` implementation.
+- `DockerBottleBackend.launch(plan)` returns a context manager
+  yielding a `Bottle` handle exposing `exec_claude(argv, *, tty=True)`,
+  `cp_in(host, ctr)`, and teardown on context exit.
+- Every existing `subprocess.run(["docker", ...])` call in
+  `cli/start.py`, `pipelock.py`, `network.py`, `ssh.py`, and
+  `skills.py` either moves into `claude_bottle/backend/docker/` or is
+  called from it. No top-level CLI code references `docker` directly.
+- `bottles[].runtime` is removed from the manifest schema, the
+  dataclass in `manifest.py`, the example manifest, and any README /
+  docs references. `require_runsc()` in the old top-level
+  `claude_bottle/docker.py` is deleted.
+- A single env var, `CLAUDE_BOTTLE_BACKEND` (default `"docker"`),
+  selects the backend. Unknown values die at startup with a list of
+  known backends.
+- The y/N preflight in `cli.py` includes the resolved Docker runtime
+  alongside the allowlist summary.
+
+## Non-goals
+
+- No second backend implementation. There is no
+  `AppleContainerBottleBackend` / `FlyioBottleBackend` in this PRD.
+  The registry in `backend/__init__.py` ships with one entry.
+- No retries, async, or streaming exec. The current code is
+  synchronous `subprocess.run`; the `Bottle` handle matches.
+- No behavior change beyond the runsc auto-detect. Pipelock topology,
+  network naming, container naming, image build flow, and SSH
+  provisioning all stay byte-identical.
+- No `--require-runsc` CLI escape hatch. If a user later wants "fail
+  rather than silently downgrade," that's a follow-up.
+- No `bottles[].backend` manifest field. Backend is a property of
+  the host environment, not the bottle definition (at least for now).
+
+## Scope
+
+### In scope
+
+- New `claude_bottle/backend/` package containing the abstract types
+  and the registry, plus a `claude_bottle/backend/docker/` subpackage
+  containing the Docker implementation.
+- The `Bottle`, `BottleBackend`, `BottlePlan`, and `BottleCleanupPlan`
+  abstract base classes; `BottleSpec` data carrier; and
+  `DockerBottleBackend` implementation.
+- Moving Docker-specific subprocess calls into the Docker subpackage.
+- Removing `bottles[].runtime` from the manifest schema and the
+  dataclass.
+- Auto-detection of `runsc` registration via `docker info`.
+- Preflight integration: the existing y/N output names the resolved
+  Docker runtime.
+- Reshaping `env.py` (formerly `env_resolve.py`) to return a
+  backend-neutral `ResolvedEnv` (`forwarded` names + `literals` map)
+  rather than writing docker-shaped files directly. The Docker
+  backend now owns the `--env-file` / `-e NAME` serialization and the
+  newline-rejection check.
+- Splitting `pipelock.py` into a backend-neutral `PipelockProxy` ABC
+  (yaml + allowlist resolution) and a `DockerPipelockProxy` subclass
+  (sidecar start/stop) under the Docker subpackage.
+- Test updates: any manifest fixtures referencing `runtime` are
+  updated; tests that assert on `--runtime=runsc` instead seed the
+  detection by mocking `docker info`.
+
+### Out of scope
+
+- Apple `container` and fly.io backends (separate PRDs, deferred
+  until the Docker backend is the only thing shipping).
+- Generalizing the pipelock sidecar to other backends. Pipelock
+  topology is, after this PRD, an implementation detail private to
+  the Docker backend.
+- Rewriting `pipelock.py`'s YAML generation. The allowlist→YAML
+  translation stays where it is and is called by the Docker backend.
+- CLI flags for runtime selection / override.
+
+## Proposed Design
+
+### New services / components
+
+A new package, `claude_bottle/backend/`, with an abstract base layer
+and a Docker subpackage:
+
+- **`claude_bottle/backend/__init__.py`** — Defines the abstract base
+  classes and the backend registry. `BottleSpec` carries the
+  CLI-supplied intent; the abstract `BottlePlan` and
+  `BottleCleanupPlan` are the prepared-but-not-launched outputs of
+  the two `prepare*` phases; `Bottle` is the running-instance handle;
+  `BottleBackend` is the dispatcher with five methods:
+
+  ```python
+  class BottleBackend(ABC):
+      name: str
+      def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> BottlePlan: ...
+      def launch(self, plan: BottlePlan) -> ContextManager[Bottle]: ...
+      def prepare_cleanup(self) -> BottleCleanupPlan: ...
+      def cleanup(self, plan: BottleCleanupPlan) -> None: ...
+      def list_active(self) -> None: ...
+  ```
+
+  The `prepare` / `launch` split lets the CLI render the y/N preflight
+  off the `BottlePlan` *before* any container or network is created.
+  The same split applies to `cleanup`. `BottleBackend.provision(plan,
+  target)` orchestrates copying skills / SSH / prompt / `.git` into a
+  running instance via four abstract sub-methods
+  (`provision_prompt`, `provision_skills`, `provision_ssh`,
+  `provision_git`); subclasses implement those four rather than
+  overriding `provision` itself.
+
+  Selection reads `CLAUDE_BOTTLE_BACKEND` (default `"docker"`).
+  Unknown values call `die()` with the list of known backends:
+
+  ```python
+  def get_bottle_backend() -> BottleBackend: ...
+  ```
+
+- **`claude_bottle/backend/docker/`** — Subpackage with the Docker
+  implementation, split into:
+  - `backend.py` — `DockerBottleBackend`, owning all five abstract
+    methods (`prepare`, `launch`, `prepare_cleanup`, `cleanup`,
+    `list_active`) plus the four `provision_*` sub-methods. Probes
+    for `runsc` availability (`docker info --format
+    '{{json .Runtimes}}'`), builds the base image and per-cwd derived
+    image, creates the per-agent internal and egress networks, brings
+    up the pipelock sidecar, runs the agent container with
+    `--runtime=runsc` iff available, copies skills / SSH keys /
+    prompt / `.git` into the running container, and tears everything
+    down on context exit.
+  - `bottle.py` — `DockerBottle`, the running-instance handle yielded
+    by `launch`.
+  - `bottle_plan.py` — `DockerBottlePlan`, the prepared-but-not-launched
+    output of `prepare`. Carries resolved container/network/image
+    names, scratch paths, and `use_runsc`. Implements `print` for the
+    y/N preflight.
+  - `bottle_cleanup_plan.py` — `DockerBottleCleanupPlan`, the analog
+    for orphan cleanup.
+  - `network.py` — Docker network helpers (create/destroy, naming).
+  - `pipelock.py` — `DockerPipelockProxy` (the sidecar start/stop
+    lifecycle) and Docker-specific naming helpers. The backend-neutral
+    yaml + allowlist resolution stays in the top-level
+    `claude_bottle/pipelock.py`.
+  - `util.py` — Docker-specific helpers (slugify, image/container
+    existence checks, `runsc_available`).
+
+### Existing code touched
+
+- **`claude_bottle/cli/start.py`** — replace the inline docker
+  orchestration with `backend = get_bottle_backend(); plan =
+  backend.prepare(spec, stage_dir=...); with backend.launch(plan) as
+  bottle: bottle.exec_claude(...)`. The y/N preflight is rendered by
+  `plan.print(...)`.
+- **`claude_bottle/manifest.py`** — drop the `runtime` field from the
+  Bottle dataclass and its validation. Existing manifests with
+  `runtime: "runsc"` produce a clear "no longer supported; gVisor is
+  now auto-detected by the backend; remove the 'runtime' field" error.
+- **`claude_bottle/docker.py`** — module deleted. `require_runsc()`,
+  `slugify()`, `image_exists()`, `container_exists()`, the
+  `build_image` / `build_image_with_cwd` helpers, and `require_docker`
+  all migrate into `claude_bottle/backend/docker/util.py` (or
+  `backend.py`).
+- **`claude_bottle/pipelock.py`** — keeps the allowlist resolution and
+  YAML generation. Becomes a thin abstract class (`PipelockProxy`)
+  exposing `prepare` (writes the yaml) plus abstract `start` / `stop`
+  methods. The Docker-specific subclass `DockerPipelockProxy` lives
+  under `backend/docker/pipelock.py`.
+- **`claude_bottle/network.py`** — folds entirely into
+  `backend/docker/network.py`. No top-level network module remains.
+- **`claude_bottle/ssh.py`** and **`claude_bottle/skills.py`** —
+  absorbed into `DockerBottleBackend` as `provision_ssh` and
+  `provision_skills`. The host-side file-tree generation stays as
+  private helpers on the backend class.
+- **`claude_bottle/env.py`** (renamed from `env_resolve.py`) —
+  `resolve_env(manifest, agent) -> ResolvedEnv` returns
+  `forwarded: list[str]` (names whose values were exported into
+  `os.environ` for inheritance) and `literals: dict[str, str]` (name
+  → verbatim value). The Docker backend translates the result into
+  `--env-file` content + `-e NAME` argv fragments.
+- **`claude_bottle/util.py`** — top-level cross-backend helpers
+  (`expand_tilde`, `is_ipv4_literal`). Backend-specific helpers live
+  in their backend's `util.py`.
+- **`claude-bottle.example.json`** — remove the `runtime` field from
+  any example bottle.
+- **`README.md`** — note `CLAUDE_BOTTLE_BACKEND` and the runsc
+  auto-detect; remove any mention of `runtime: "runsc"` as a manifest
+  field.
+
+### Data model changes
+
+The bottle schema loses one field:
+
+```diff
+ {
+   "bottles": {
+     "default": {
+-      "runtime": "runsc",
+       "env":  { "...": "..." },
+       "ssh":  [],
+       "egress": { "allowlist": [...] }
+     }
+   }
+ }
+```
+
+Any manifest carrying `runtime` produces a validation error on load
+(`"bottle '<name>' has a 'runtime' field, which is no longer
+supported. gVisor (runsc) is now auto-detected by the backend;
+remove the 'runtime' field from the bottle definition."`).
+
+The agent schema is unchanged.
+
+### External dependencies
+
+None new. This PRD reorganizes existing code; it does not pull in any
+new images, binaries, or libraries.
+
+### Behavior the runsc auto-detect introduces
+
+`DockerBottleBackend.prepare` runs `docker info --format
+'{{json .Runtimes}}'` exactly once per call. If `runsc` is in the
+output, `use_runsc` is set on the `DockerBottlePlan` and the
+subsequent `docker run` adds `--runtime=runsc`. Otherwise it runs
+without that flag. The choice is logged via the existing `info()`
+helper as part of the preflight:
+
+```
+docker runtime: runsc (gVisor)        # or: runc (default)
+```
+
+The y/N preflight (rendered by `DockerBottlePlan.print`) shows the
+same line, so users can confirm what they're about to run under
+before approving.
+
+## References
+
+- `docs/research/apple-container-backend.md` — original motivation;
+  prior draft considered a low-level `Backend` protocol and rejected
+  it as the wrong layer.
+- `docs/research/bash-vs-python-vs-go.md` §Recommendation — argues
+  that the backend abstraction matters independent of language choice.
+- PRD 0001 (`docs/prds/0001-per-agent-egress-proxy-via-pipelock.md`)
+  — defines the pipelock topology that becomes a private
+  implementation detail of the Docker backend after this PRD ships.

tests/test_manifest_runtime.py

@@ -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__":

tests/test_orphan_cleanup.py

@@ -1,18 +1,22 @@
 """Integration: the cleanup primitives the start-flow trap depends on
 are idempotent. The original orphan-network bug was a trap-ordering
 issue; the fix moved the install earlier. The trap is only safe if
-network_remove and pipelock_stop are no-ops against missing resources."""
+network_remove and PipelockProxy.stop are no-ops against missing
+resources."""
 
 import os
 import subprocess
 import unittest
 
-from claude_bottle.network import (
+from claude_bottle.backend.docker.network import (
     network_create_egress,
     network_create_internal,
     network_remove,
 )
-from claude_bottle.pipelock import pipelock_stop
+from claude_bottle.backend.docker.pipelock import (
+    DockerPipelockProxy,
+    pipelock_container_name,
+)
 from tests._docker import skip_unless_docker
 
 
@@ -68,7 +72,7 @@ class TestOrphanCleanup(unittest.TestCase):
 
     def test_pipelock_stop_missing_sidecar(self):
         # Should not raise.
-        pipelock_stop(f"missing-{self.slug}")
+        DockerPipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
 
 
 if __name__ == "__main__":

tests/test_pipelock_allowlist.py

@@ -18,13 +18,13 @@ from tests.fixtures import fixture_minimal, fixture_with_egress, fixture_with_ss
 
 class TestBottleAllowlist(unittest.TestCase):
     def test_egress_allowlist_present(self):
-        out = pipelock_bottle_allowlist(fixture_with_egress(), "dev")
+        out = pipelock_bottle_allowlist(fixture_with_egress().bottles["dev"])
         self.assertIn("github.com", out)
         self.assertIn("gitlab.com", out)
         self.assertIn("registry.npmjs.org", out)
 
     def test_empty_when_no_egress_block(self):
-        out = pipelock_bottle_allowlist(fixture_minimal(), "dev")
+        out = pipelock_bottle_allowlist(fixture_minimal().bottles["dev"])
         self.assertEqual([], out)
 
     def test_rejects_non_string_entry(self):
@@ -38,17 +38,17 @@ class TestBottleAllowlist(unittest.TestCase):
 
 class TestSSHHostnames(unittest.TestCase):
     def test_hostnames_include_both(self):
-        hosts = pipelock_bottle_ssh_hostnames(fixture_with_ssh(), "dev")
+        hosts = pipelock_bottle_ssh_hostnames(fixture_with_ssh().bottles["dev"])
         self.assertIn("100.78.141.42", hosts)
         self.assertIn("github.com", hosts)
 
     def test_ip_cidrs_only_ipv4(self):
-        cidrs = pipelock_bottle_ssh_ip_cidrs(fixture_with_ssh(), "dev")
+        cidrs = pipelock_bottle_ssh_ip_cidrs(fixture_with_ssh().bottles["dev"])
         self.assertIn("100.78.141.42/32", cidrs)
         self.assertNotIn("github.com", cidrs)
 
     def test_trusted_domains_only_hostnames(self):
-        trusted = pipelock_bottle_ssh_trusted_domains(fixture_with_ssh(), "dev")
+        trusted = pipelock_bottle_ssh_trusted_domains(fixture_with_ssh().bottles["dev"])
         self.assertIn("github.com", trusted)
         self.assertNotIn("100.78.141.42", trusted)
 
@@ -69,7 +69,7 @@ class TestEffectiveAllowlist(unittest.TestCase):
             },
             "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
         })
-        eff = pipelock_effective_allowlist(manifest, "dev")
+        eff = pipelock_effective_allowlist(manifest.bottles["dev"])
         self.assertIn("api.anthropic.com", eff)
         self.assertIn("registry.npmjs.org", eff)
         self.assertIn("100.78.141.42", eff)

tests/test_pipelock_classify.py

@@ -1,10 +1,10 @@
 """Unit: is_ipv4_literal — the classifier that decides whether
-bottle.ssh[].Hostname goes into ssrf.ip_allowlist (IPv4 literal) or
-trusted_domains (hostname)."""
+bottle.ssh[].Hostname goes into pipelock's ssrf.ip_allowlist (IPv4
+literal) or trusted_domains (hostname)."""
 
 import unittest
 
-from claude_bottle.pipelock import is_ipv4_literal
+from claude_bottle.util import is_ipv4_literal
 
 
 class TestIPv4Classify(unittest.TestCase):

tests/test_pipelock_image.py

@@ -1,15 +1,11 @@
-"""Integration: verify the pinned pipelock image. Requires docker.
-  - Pinned digest is reachable on the registry.
-  - Image's ENTRYPOINT/CMD match what claude_bottle.pipelock assumes
-    (`/pipelock` and `run --listen 0.0.0.0:8888`).
-  - The /pipelock binary actually runs (--version succeeds)."""
+"""Integration: the pinned pipelock image's binary actually runs.
+Catches a broken upstream packaging at the pinned digest. Requires
+docker."""
 
-import json
-import re
 import subprocess
 import unittest
 
-from claude_bottle.pipelock import PIPELOCK_IMAGE
+from claude_bottle.backend.docker.pipelock import PIPELOCK_IMAGE
 from tests._docker import skip_unless_docker
 
 
@@ -17,7 +13,6 @@ from tests._docker import skip_unless_docker
 class TestPipelockImage(unittest.TestCase):
     @classmethod
     def setUpClass(cls):
-        # Pull the pinned image (cheap if cached).
         result = subprocess.run(
             ["docker", "pull", PIPELOCK_IMAGE],
             stdout=subprocess.DEVNULL,
@@ -26,22 +21,6 @@ class TestPipelockImage(unittest.TestCase):
         if result.returncode != 0:
             raise unittest.SkipTest(f"could not pull {PIPELOCK_IMAGE}")
 
-    def test_entrypoint_contains_pipelock(self):
-        result = subprocess.run(
-            ["docker", "image", "inspect", PIPELOCK_IMAGE,
-             "--format", "{{json .Config.Entrypoint}}"],
-            capture_output=True, text=True,
-        )
-        self.assertIn("/pipelock", result.stdout)
-
-    def test_cmd_contains_run(self):
-        result = subprocess.run(
-            ["docker", "image", "inspect", PIPELOCK_IMAGE,
-             "--format", "{{json .Config.Cmd}}"],
-            capture_output=True, text=True,
-        )
-        self.assertIn("run", result.stdout)
-
     def test_binary_runs(self):
         result = subprocess.run(
             ["docker", "run", "--rm", PIPELOCK_IMAGE, "--version"],

tests/test_pipelock_naming.py

@@ -1,33 +0,0 @@
-"""Unit: pipelock naming helpers (container_name, proxy_url, proxy_host_port)."""
-
-import unittest
-
-from claude_bottle.pipelock import (
-    pipelock_container_name,
-    pipelock_proxy_host_port,
-    pipelock_proxy_url,
-)
-
-
-class TestPipelockNaming(unittest.TestCase):
-    def test_container_name_simple(self):
-        self.assertEqual("claude-bottle-pipelock-foo", pipelock_container_name("foo"))
-
-    def test_container_name_with_hyphens(self):
-        self.assertEqual(
-            "claude-bottle-pipelock-some-slug", pipelock_container_name("some-slug")
-        )
-
-    def test_proxy_url_default_port(self):
-        self.assertEqual(
-            "http://claude-bottle-pipelock-foo:8888", pipelock_proxy_url("foo")
-        )
-
-    def test_proxy_host_port_default_port(self):
-        self.assertEqual(
-            "claude-bottle-pipelock-foo:8888", pipelock_proxy_host_port("foo")
-        )
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/test_pipelock_sidecar_smoke.py

@@ -12,7 +12,10 @@ import unittest
 import urllib.request
 from pathlib import Path
 
-from claude_bottle.pipelock import PIPELOCK_IMAGE, pipelock_write_yaml
+from claude_bottle.backend.docker.pipelock import (
+    PIPELOCK_IMAGE,
+    DockerPipelockProxy,
+)
 from tests._docker import skip_unless_docker
 from tests.fixtures import fixture_minimal
 
@@ -38,7 +41,7 @@ class TestPipelockSidecarSmoke(unittest.TestCase):
     )
     def test_smoke(self):
         yaml_path = self.work_dir / "pipelock.yaml"
-        pipelock_write_yaml(fixture_minimal(), "dev", yaml_path)
+        DockerPipelockProxy().prepare(fixture_minimal().bottles["dev"], "demo", yaml_path)
 
         create = subprocess.run(
             [

tests/test_pipelock_yaml.py

@@ -1,20 +1,21 @@
-"""Unit: pipelock_write_yaml — produces a YAML config containing the
-expected top-level keys and per-bottle entries. We don't fully parse
-YAML; we grep for content shape."""
+"""Unit: PipelockProxy.prepare — produces a pipelock YAML config
+containing the expected top-level keys and per-bottle entries. We
+don't fully parse YAML; we grep for content shape."""
 
 import os
 import tempfile
 import unittest
 from pathlib import Path
 
+from claude_bottle.backend.docker.pipelock import DockerPipelockProxy
 from claude_bottle.manifest import Manifest
-from claude_bottle.pipelock import pipelock_write_yaml
 from tests.fixtures import fixture_minimal, fixture_with_ssh
 
 
-class TestPipelockYaml(unittest.TestCase):
+class TestPipelockProxyPrepare(unittest.TestCase):
     def setUp(self):
         self.out_dir = Path(tempfile.mkdtemp())
+        self.proxy = DockerPipelockProxy()
 
     def tearDown(self):
         import shutil
@@ -22,7 +23,7 @@ class TestPipelockYaml(unittest.TestCase):
 
     def test_minimal(self):
         yaml_path = self.out_dir / "min.yaml"
-        pipelock_write_yaml(fixture_minimal(), "dev", yaml_path)
+        self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path)
         content = yaml_path.read_text()
         self.assertIn("mode: strict", content)
         self.assertIn("enforce: true", content)
@@ -40,7 +41,7 @@ class TestPipelockYaml(unittest.TestCase):
 
     def test_ssh_blocks(self):
         yaml_path = self.out_dir / "ssh.yaml"
-        pipelock_write_yaml(fixture_with_ssh(), "dev", yaml_path)
+        self.proxy.prepare(fixture_with_ssh().bottles["dev"], "demo", yaml_path)
         content = yaml_path.read_text()
         self.assertIn("trusted_domains:", content)
         self.assertIn("github.com", content)
@@ -64,7 +65,7 @@ class TestPipelockYaml(unittest.TestCase):
             "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
         })
         yaml_path = self.out_dir / "secret.yaml"
-        pipelock_write_yaml(manifest, "dev", yaml_path)
+        self.proxy.prepare(manifest.bottles["dev"], "demo", yaml_path)
         content = yaml_path.read_text()
         self.assertNotIn("literal-value-should-not-appear", content)
         self.assertNotIn("MY_SECRET", content)
@@ -72,7 +73,7 @@ class TestPipelockYaml(unittest.TestCase):
 
     def test_file_mode_is_600(self):
         yaml_path = self.out_dir / "min.yaml"
-        pipelock_write_yaml(fixture_minimal(), "dev", yaml_path)
+        self.proxy.prepare(fixture_minimal().bottles["dev"], "demo", yaml_path)
         mode = os.stat(yaml_path).st_mode & 0o777
         self.assertEqual(0o600, mode)