refactor(pipelock): start/stop become methods on PipelockProxy

ProxyPlan -> PipelockProxyPlan, with two additional fields populated
in launch: internal_network, egress_network (default ""). prepare
fills yaml_path + slug; launch uses dataclasses.replace to populate
the networks before calling start.

pipelock_start -> PipelockProxy.start(plan). Reads yaml_path, slug,
internal_network, egress_network off the plan. Returns the resolved
container name.

pipelock_stop -> PipelockProxy.stop(proxy_target). Takes the resolved
container name directly (the value that start returned); no longer
needs to know about slugs or naming conventions.

Backend launch passes the running container name (state["pipelock"])
to stop. Test for stop's idempotency uses pipelock_container_name to
construct the proxy_target.

claude_bottle/backend/docker/backend.py

@@ -10,6 +10,7 @@ Methods:
 
 from __future__ import annotations
 
+import dataclasses
 import os
 import subprocess
 import sys
@@ -125,15 +126,15 @@ class DockerBottleBackend(BottleBackend):
             use_runsc=use_runsc,
         )
 
-    def prepare_proxy(self, spec: BottleSpec, stage_dir: Path) -> pipelock.ProxyPlan:
+    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 ProxyPlan
-        for the launch step to consume. Stage-only: no Docker resources
-        created yet."""
+        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_name = spec.manifest.agents[spec.agent_name].bottle
-        self._proxy.prepare(spec.manifest, bottle_name, yaml_path)
-        return pipelock.ProxyPlan(yaml_path=yaml_path)
+        slug = docker_mod.slugify(spec.agent_name)
+        return self._proxy.prepare(spec.manifest, bottle_name, slug, yaml_path)
 
     @contextmanager
     def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]:
@@ -160,7 +161,7 @@ class DockerBottleBackend(BottleBackend):
                     )
                     state["container"] = ""
                 if state["pipelock"]:
-                    pipelock.pipelock_stop(plan.slug)
+                    self._proxy.stop(state["pipelock"])
                     state["pipelock"] = ""
                 if state["internal_network"]:
                     network_mod.network_remove(state["internal_network"])
@@ -182,13 +183,12 @@ class DockerBottleBackend(BottleBackend):
 
             state["internal_network"] = network_mod.network_create_internal(plan.slug)
             state["egress_network"] = network_mod.network_create_egress(plan.slug)
-            state["pipelock"] = pipelock.pipelock_start(
-                plan.slug,
-                state["internal_network"],
-                state["egress_network"],
-                plan.proxy_plan.yaml_path.parent,
-                plan.proxy_plan.yaml_path.name,
+            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

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 ProxyPlan
+from ...pipelock import PipelockProxyPlan
 from .. import BottlePlan
 
 
@@ -31,7 +31,7 @@ class DockerBottlePlan(BottlePlan):
     env_file: Path
     args_file: Path
     prompt_file: Path
-    proxy_plan: ProxyPlan
+    proxy_plan: PipelockProxyPlan
     allowlist_summary: str
     use_runsc: bool
 

claude_bottle/pipelock.py

@@ -125,21 +125,28 @@ def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str:
 
 
 @dataclass(frozen=True)
-class ProxyPlan:
-    """Output of a proxy's prepare step; consumed by launch when the
-    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."""
+class PipelockProxyPlan:
+    """Output of PipelockProxy.prepare; consumed by .start when the
+    sidecar needs to be brought up.
+
+    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 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."""
+    generation and the sidecar's start/stop lifecycle."""
 
-    def prepare(self, manifest: Manifest, bottle_name: str, yaml_path: Path) -> None:
+    def prepare(
+        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
@@ -183,82 +190,80 @@ class PipelockProxy:
         yaml_path.write_text("\n".join(lines) + "\n")
         yaml_path.chmod(0o600)
 
+        return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
 
-# --- Sidecar lifecycle -----------------------------------------------------
+    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}")
 
-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}; backend.prepare_proxy must run first")
+        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}")
 
-    info(f"starting pipelock sidecar {name} on network {internal_network}")
+        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()}")
 
-    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],
+            ["docker", "network", "connect", plan.egress_network, name],
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
         ).returncode != 0:
-            warn(f"failed to remove pipelock sidecar {name}; clean up with 'docker rm -f {name}'")
+            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}'"
+                )

tests/test_orphan_cleanup.py

@@ -1,7 +1,8 @@
 """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
@@ -12,7 +13,7 @@ from claude_bottle.backend.docker.network import (
     network_create_internal,
     network_remove,
 )
-from claude_bottle.pipelock import pipelock_stop
+from claude_bottle.pipelock import PipelockProxy, pipelock_container_name
 from tests._docker import skip_unless_docker
 
 
@@ -68,7 +69,7 @@ class TestOrphanCleanup(unittest.TestCase):
 
     def test_pipelock_stop_missing_sidecar(self):
         # Should not raise.
-        pipelock_stop(f"missing-{self.slug}")
+        PipelockProxy().stop(pipelock_container_name(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", yaml_path)
+        PipelockProxy().prepare(fixture_minimal(), "dev", "demo", 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", yaml_path)
+        self.proxy.prepare(fixture_minimal(), "dev", "demo", 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", yaml_path)
+        self.proxy.prepare(fixture_with_ssh(), "dev", "demo", 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", yaml_path)
+        self.proxy.prepare(manifest, "dev", "demo", 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", yaml_path)
+        self.proxy.prepare(fixture_minimal(), "dev", "demo", yaml_path)
         mode = os.stat(yaml_path).st_mode & 0o777
         self.assertEqual(0o600, mode)