refactor(pipelock): prepare_proxy returns a ProxyPlan

Add a frozen ProxyPlan dataclass to pipelock.py (currently one field:
yaml_path; kept as a class so future proxy-level state has a home).

  - prepare_proxy(spec, stage_dir) now returns pipelock.ProxyPlan
    instead of a raw Path.
  - DockerBottlePlan replaces pipelock_yaml_path + pipelock_yaml_filename
    with a single proxy: ProxyPlan field.
  - launch reads plan.proxy.yaml_path.parent / .name when calling
    pipelock_start. Eventually pipelock_start should just take a Path
    but that's a separate change.

claude_bottle/backend/docker/backend.py

@@ -10,7 +10,6 @@ Methods:
 
 from __future__ import annotations
 
-import dataclasses
 import os
 import subprocess
 import sys
@@ -121,20 +120,20 @@ class DockerBottleBackend(BottleBackend):
             env_file=env_file,
             args_file=args_file,
             prompt_file=prompt_file,
-            proxy_plan=proxy_plan,
+            proxy=proxy_plan,
             allowlist_summary=allowlist_summary,
             use_runsc=use_runsc,
         )
 
-    def prepare_proxy(self, spec: BottleSpec, stage_dir: Path) -> pipelock.PipelockProxyPlan:
+    def prepare_proxy(self, spec: BottleSpec, stage_dir: Path) -> pipelock.ProxyPlan:
         """Decide where the pipelock yaml lives in `stage_dir`, delegate
-        to PipelockProxy to write it, and return the resolved
+        to PipelockProxy to write it, and return the resolved ProxyPlan
-        PipelockProxyPlan for the launch step to consume. Stage-only:
+        for the launch step to consume. Stage-only: no Docker resources
-        no Docker resources created yet."""
+        created yet."""
         yaml_path = stage_dir / "pipelock.yaml"
         bottle_name = spec.manifest.agents[spec.agent_name].bottle
-        slug = docker_mod.slugify(spec.agent_name)
+        self._proxy.prepare(spec.manifest, bottle_name, yaml_path)
-        return self._proxy.prepare(spec.manifest, bottle_name, slug, yaml_path)
+        return pipelock.ProxyPlan(yaml_path=yaml_path)
 
     @contextmanager
     def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]:
@@ -161,7 +160,7 @@ class DockerBottleBackend(BottleBackend):
                     )
                     state["container"] = ""
                 if state["pipelock"]:
-                    self._proxy.stop(state["pipelock"])
+                    pipelock.pipelock_stop(plan.slug)
                     state["pipelock"] = ""
                 if state["internal_network"]:
                     network_mod.network_remove(state["internal_network"])
@@ -183,12 +182,13 @@ class DockerBottleBackend(BottleBackend):
 
             state["internal_network"] = network_mod.network_create_internal(plan.slug)
             state["egress_network"] = network_mod.network_create_egress(plan.slug)
-            proxy_plan = dataclasses.replace(
+            state["pipelock"] = pipelock.pipelock_start(
-                plan.proxy_plan,
+                plan.slug,
-                internal_network=state["internal_network"],
+                state["internal_network"],
-                egress_network=state["egress_network"],
+                state["egress_network"],
+                plan.proxy.yaml_path.parent,
+                plan.proxy.yaml_path.name,
             )
-            state["pipelock"] = self._proxy.start(proxy_plan)
 
             container = self._run_agent_container(plan, state["internal_network"])
             state["container"] = container

claude_bottle/backend/docker/bottle_plan.py

@@ -12,7 +12,7 @@ from dataclasses import dataclass
 from pathlib import Path
 
 from ...log import info
-from ...pipelock import PipelockProxyPlan
+from ...pipelock import ProxyPlan
 from .. import BottlePlan
 
 
@@ -31,7 +31,7 @@ class DockerBottlePlan(BottlePlan):
     env_file: Path
     args_file: Path
     prompt_file: Path
-    proxy_plan: PipelockProxyPlan
+    proxy: ProxyPlan
     allowlist_summary: str
     use_runsc: bool
 

claude_bottle/pipelock.py

@@ -125,28 +125,21 @@ def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str:
 
 
 @dataclass(frozen=True)
-class PipelockProxyPlan:
+class ProxyPlan:
-    """Output of PipelockProxy.prepare; consumed by .start when the
+    """Output of a proxy's prepare step; consumed by launch when the
-    sidecar needs to be brought up.
+    proxy needs to be brought up. Currently single-field (the on-host
-
+    yaml path); kept as a dataclass so future proxy state has a home."""
-    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."""
 
     yaml_path: Path
-    slug: str
-    internal_network: str = ""
-    egress_network: str = ""
 
 
 class PipelockProxy:
     """The pipelock egress proxy. Encapsulates the YAML-config
-    generation and the sidecar's start/stop lifecycle."""
+    generation (and is the natural home for any future proxy-level
+    state). Backends that use pipelock hold a PipelockProxy instance
+    and delegate the prepare step to it."""
 
-    def prepare(
+    def prepare(self, manifest: Manifest, bottle_name: str, yaml_path: Path) -> None:
-        self, manifest: Manifest, bottle_name: str, slug: str, yaml_path: Path
-    ) -> PipelockProxyPlan:
         """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
@@ -190,80 +183,82 @@ class PipelockProxy:
         yaml_path.write_text("\n".join(lines) + "\n")
         yaml_path.chmod(0o600)
 
-        return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
 
-    def start(self, plan: PipelockProxyPlan) -> str:
+# --- Sidecar lifecycle -----------------------------------------------------
-        """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 = [
+def pipelock_start(
-            "docker", "create",
+    slug: str,
-            "--name", name,
+    internal_network: str,
-            "--network", plan.internal_network,
+    egress_network: str,
-            PIPELOCK_IMAGE,
+    yaml_dir: Path,
-            "run", "--config", "/etc/pipelock.yaml",
+    yaml_filename: str,
-            "--listen", f"0.0.0.0:{PIPELOCK_PORT}",
+) -> str:
-        ]
+    """Boot the pipelock sidecar:
-        if subprocess.run(create_args, stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL).returncode != 0:
+      1. `docker create` on the internal network with the canonical name
-            die(f"failed to create pipelock sidecar {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}; backend.prepare_proxy must run first")
 
-        cp_result = subprocess.run(
+    info(f"starting pipelock sidecar {name} on network {internal_network}")
-            ["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()}")
 
+    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", "network", "connect", plan.egress_network, name],
+            ["docker", "rm", "-f", name],
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
         ).returncode != 0:
-            subprocess.run(["docker", "rm", "-f", name], stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL)
+            warn(f"failed to remove pipelock sidecar {name}; clean up with 'docker rm -f {name}'")
-            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}'"
-                )

tests/test_orphan_cleanup.py

@@ -1,8 +1,7 @@
 """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 PipelockProxy.stop are no-ops against missing
+network_remove and pipelock_stop are no-ops against missing resources."""
-resources."""
 
 import os
 import subprocess
@@ -13,7 +12,7 @@ from claude_bottle.backend.docker.network import (
     network_create_internal,
     network_remove,
 )
-from claude_bottle.pipelock import PipelockProxy, pipelock_container_name
+from claude_bottle.pipelock import pipelock_stop
 from tests._docker import skip_unless_docker
 
 
@@ -69,7 +68,7 @@ class TestOrphanCleanup(unittest.TestCase):
 
     def test_pipelock_stop_missing_sidecar(self):
         # Should not raise.
-        PipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
+        pipelock_stop(f"missing-{self.slug}")
 
 
 if __name__ == "__main__":

tests/test_pipelock_sidecar_smoke.py

@@ -38,7 +38,7 @@ class TestPipelockSidecarSmoke(unittest.TestCase):
     )
     def test_smoke(self):
         yaml_path = self.work_dir / "pipelock.yaml"
-        PipelockProxy().prepare(fixture_minimal(), "dev", "demo", yaml_path)
+        PipelockProxy().prepare(fixture_minimal(), "dev", yaml_path)
 
         create = subprocess.run(
             [

tests/test_pipelock_yaml.py

@@ -23,7 +23,7 @@ class TestPipelockProxyPrepare(unittest.TestCase):
 
     def test_minimal(self):
         yaml_path = self.out_dir / "min.yaml"
-        self.proxy.prepare(fixture_minimal(), "dev", "demo", yaml_path)
+        self.proxy.prepare(fixture_minimal(), "dev", yaml_path)
         content = yaml_path.read_text()
         self.assertIn("mode: strict", content)
         self.assertIn("enforce: true", content)
@@ -41,7 +41,7 @@ class TestPipelockProxyPrepare(unittest.TestCase):
 
     def test_ssh_blocks(self):
         yaml_path = self.out_dir / "ssh.yaml"
-        self.proxy.prepare(fixture_with_ssh(), "dev", "demo", yaml_path)
+        self.proxy.prepare(fixture_with_ssh(), "dev", yaml_path)
         content = yaml_path.read_text()
         self.assertIn("trusted_domains:", content)
         self.assertIn("github.com", content)
@@ -65,7 +65,7 @@ class TestPipelockProxyPrepare(unittest.TestCase):
             "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
         })
         yaml_path = self.out_dir / "secret.yaml"
-        self.proxy.prepare(manifest, "dev", "demo", yaml_path)
+        self.proxy.prepare(manifest, "dev", yaml_path)
         content = yaml_path.read_text()
         self.assertNotIn("literal-value-should-not-appear", content)
         self.assertNotIn("MY_SECRET", content)
@@ -73,7 +73,7 @@ class TestPipelockProxyPrepare(unittest.TestCase):
 
     def test_file_mode_is_600(self):
         yaml_path = self.out_dir / "min.yaml"
-        self.proxy.prepare(fixture_minimal(), "dev", "demo", yaml_path)
+        self.proxy.prepare(fixture_minimal(), "dev", yaml_path)
         mode = os.stat(yaml_path).st_mode & 0o777
         self.assertEqual(0o600, mode)