refactor(pipelock): introduce PipelockProxy class housing the yaml body

The YAML generation now lives on PipelockProxy.prepare(manifest,
bottle_name, yaml_path) in claude_bottle/pipelock.py. The class is the
natural home for any future proxy-level state.

DockerBottleBackend keeps an instance as a class attribute
(_proxy = PipelockProxy()) and its prepare_proxy becomes a thin
delegation. A future backend that wants a different egress proxy
(or none) plugs in its own strategy.

Tests retarget at the new home — PipelockProxy.prepare gets the
content-shape assertions; the sidecar smoke test uses the class
directly too. Same coverage.

claude_bottle/backend/docker/backend.py

@@ -39,6 +39,7 @@ class DockerBottleBackend(BottleBackend):
     (default)."""
 
     name = "docker"
+    _proxy: pipelock.PipelockProxy = pipelock.PipelockProxy()
 
     def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
         """Resolve names, validate, write scratch files. No Docker
@@ -128,49 +129,10 @@ class DockerBottleBackend(BottleBackend):
         )
 
     def prepare_proxy(self, spec: BottleSpec, yaml_path: Path) -> None:
-        """Write the pipelock proxy's yaml config (mode 600) to
-        `yaml_path` for the sidecar to consume when it boots in
-        `launch`. 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."""
+        """Delegate to PipelockProxy to write the sidecar's yaml
+        config. Stage-only: no Docker resources created yet."""
         bottle_name = spec.manifest.agents[spec.agent_name].bottle
-        allowlist = pipelock.pipelock_effective_allowlist(spec.manifest, bottle_name)
-        trusted = pipelock.pipelock_bottle_ssh_trusted_domains(spec.manifest, bottle_name)
-        ip_cidrs = pipelock.pipelock_bottle_ssh_ip_cidrs(spec.manifest, bottle_name)
-
-        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}"')
-            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")
-
-        yaml_path.write_text("\n".join(lines) + "\n")
-        yaml_path.chmod(0o600)
+        self._proxy.prepare(spec.manifest, bottle_name, yaml_path)
 
     @contextmanager
     def launch(self, plan: BottlePlan) -> Iterator[DockerBottle]:

claude_bottle/pipelock.py

@@ -120,6 +120,60 @@ def pipelock_allowlist_summary(manifest: Manifest, bottle_name: str) -> str:
 
 
 
+# --- Proxy class -----------------------------------------------------------
+
+
+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."""
+
+    def prepare(self, manifest: Manifest, bottle_name: str, yaml_path: Path) -> None:
+        """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(manifest, bottle_name)
+        trusted = pipelock_bottle_ssh_trusted_domains(manifest, bottle_name)
+        ip_cidrs = pipelock_bottle_ssh_ip_cidrs(manifest, bottle_name)
+
+        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}"')
+            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")
+
+        yaml_path.write_text("\n".join(lines) + "\n")
+        yaml_path.chmod(0o600)
+
+
 # --- Sidecar lifecycle -----------------------------------------------------
 
 

tests/test_pipelock_sidecar_smoke.py

@@ -12,9 +12,7 @@ import unittest
 import urllib.request
 from pathlib import Path
 
-from claude_bottle.backend import BottleSpec
-from claude_bottle.backend.docker import DockerBottleBackend
-from claude_bottle.pipelock import PIPELOCK_IMAGE
+from claude_bottle.pipelock import PIPELOCK_IMAGE, PipelockProxy
 from tests._docker import skip_unless_docker
 from tests.fixtures import fixture_minimal
 
@@ -40,14 +38,7 @@ class TestPipelockSidecarSmoke(unittest.TestCase):
     )
     def test_smoke(self):
         yaml_path = self.work_dir / "pipelock.yaml"
-        spec = BottleSpec(
-            manifest=fixture_minimal(),
-            agent_name="demo",
-            copy_cwd=False,
-            user_cwd="/tmp",
-            forward_oauth_token=False,
-        )
-        DockerBottleBackend().prepare_proxy(spec, yaml_path)
+        PipelockProxy().prepare(fixture_minimal(), "dev", yaml_path)
 
         create = subprocess.run(
             [

tests/test_pipelock_yaml.py

@@ -1,34 +1,21 @@
-"""Unit: DockerBottleBackend.prepare_proxy — 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."""
+"""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 import BottleSpec
-from claude_bottle.backend.docker import DockerBottleBackend
 from claude_bottle.manifest import Manifest
+from claude_bottle.pipelock import PipelockProxy
 from tests.fixtures import fixture_minimal, fixture_with_ssh
 
 
-def _spec(manifest: Manifest) -> BottleSpec:
-    """Construct a minimal BottleSpec around a fixture manifest. The
-    fixtures all define an agent named 'demo' on a bottle named 'dev'."""
-    return BottleSpec(
-        manifest=manifest,
-        agent_name="demo",
-        copy_cwd=False,
-        user_cwd="/tmp",
-        forward_oauth_token=False,
-    )
-
-
-class TestPrepareProxyYaml(unittest.TestCase):
+class TestPipelockProxyPrepare(unittest.TestCase):
     def setUp(self):
         self.out_dir = Path(tempfile.mkdtemp())
-        self.backend = DockerBottleBackend()
+        self.proxy = PipelockProxy()
 
     def tearDown(self):
         import shutil
@@ -36,7 +23,7 @@ class TestPrepareProxyYaml(unittest.TestCase):
 
     def test_minimal(self):
         yaml_path = self.out_dir / "min.yaml"
-        self.backend.prepare_proxy(_spec(fixture_minimal()), 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)
@@ -54,7 +41,7 @@ class TestPrepareProxyYaml(unittest.TestCase):
 
     def test_ssh_blocks(self):
         yaml_path = self.out_dir / "ssh.yaml"
-        self.backend.prepare_proxy(_spec(fixture_with_ssh()), 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)
@@ -78,7 +65,7 @@ class TestPrepareProxyYaml(unittest.TestCase):
             "agents": {"demo": {"skills": [], "prompt": "", "bottle": "dev"}},
         })
         yaml_path = self.out_dir / "secret.yaml"
-        self.backend.prepare_proxy(_spec(manifest), 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)
@@ -86,7 +73,7 @@ class TestPrepareProxyYaml(unittest.TestCase):
 
     def test_file_mode_is_600(self):
         yaml_path = self.out_dir / "min.yaml"
-        self.backend.prepare_proxy(_spec(fixture_minimal()), yaml_path)
+        self.proxy.prepare(fixture_minimal(), "dev", yaml_path)
         mode = os.stat(yaml_path).st_mode & 0o777
         self.assertEqual(0o600, mode)