refactor(pipelock): split sidecar lifecycle into DockerPipelockProxy

PipelockProxy becomes an ABC with the platform-agnostic
prepare/_build_pipelock_yaml as concrete methods and start/stop as
abstract. Docker-specific sidecar lifecycle moves to a new sibling
file:

  claude_bottle/backend/docker/pipelock.py
    DockerPipelockProxy(PipelockProxy) — implements start (docker
    create/cp/network connect/start) and stop (docker inspect/rm -f).

DockerBottleBackend._proxy is now a DockerPipelockProxy instance.
Tests that previously instantiated PipelockProxy() directly switch to
DockerPipelockProxy() (the base is no longer constructable).

claude_bottle/backend/docker/backend.py

@@ -29,6 +29,7 @@ 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
 
 
 # Where the repo root lives, for `docker build` context. Computed once.
@@ -40,7 +41,7 @@ class DockerBottleBackend(BottleBackend):
     (default)."""
 
     name = "docker"
-    _proxy: pipelock.PipelockProxy = pipelock.PipelockProxy()
+    _proxy: DockerPipelockProxy = DockerPipelockProxy()
 
     def prepare(self, spec: BottleSpec, *, stage_dir: Path) -> DockerBottlePlan:
         """Resolve names, validate, write scratch files. No Docker

claude_bottle/backend/docker/pipelock.py

@@ -0,0 +1,96 @@
+"""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 subprocess
+
+from ...log import die, info, warn
+from ...pipelock import (
+    PIPELOCK_IMAGE,
+    PIPELOCK_PORT,
+    PipelockProxy,
+    PipelockProxyPlan,
+    pipelock_container_name,
+)
+
+
+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/pipelock.py

@@ -13,11 +13,10 @@ Image pin: ghcr.io/luckypipewrench/pipelock@sha256:<digest> for tag 2.3.0.
 from __future__ import annotations
 
 import os
-import subprocess
+from abc import ABC, abstractmethod
 from dataclasses import dataclass
 from pathlib import Path
 
-from .log import die, info, warn
 from .manifest import Bottle, Manifest
 from .util import is_ipv4_literal
 
@@ -128,9 +127,10 @@ class PipelockProxyPlan:
     egress_network: str = ""
 
 
-class PipelockProxy:
+class PipelockProxy(ABC):
     """The pipelock egress proxy. Encapsulates the YAML-config
-    generation and the sidecar's start/stop lifecycle."""
+    generation; the sidecar's start/stop lifecycle is backend-specific
+    and lives on concrete subclasses (e.g. DockerPipelockProxy)."""
 
     def prepare(
         self, manifest: Manifest, bottle_name: str, slug: str, yaml_path: Path
@@ -186,78 +186,14 @@ class PipelockProxy:
 
         return PipelockProxyPlan(yaml_path=yaml_path, slug=slug)
 
+    @abstractmethod
     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
+        """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."""
 
+    @abstractmethod
     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}'"
-                )
+        """Tear down the pipelock sidecar identified by `proxy_target`
+        (the value `.start` returned). Idempotent: a missing target is
+        success. Backend-specific."""

tests/test_orphan_cleanup.py

@@ -13,7 +13,8 @@ from claude_bottle.backend.docker.network import (
     network_create_internal,
     network_remove,
 )
-from claude_bottle.pipelock import PipelockProxy, pipelock_container_name
+from claude_bottle.backend.docker.pipelock import DockerPipelockProxy
+from claude_bottle.pipelock import pipelock_container_name
 from tests._docker import skip_unless_docker
 
 
@@ -69,7 +70,7 @@ class TestOrphanCleanup(unittest.TestCase):
 
     def test_pipelock_stop_missing_sidecar(self):
         # Should not raise.
-        PipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
+        DockerPipelockProxy().stop(pipelock_container_name(f"missing-{self.slug}"))
 
 
 if __name__ == "__main__":

tests/test_pipelock_sidecar_smoke.py

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

tests/test_pipelock_yaml.py

@@ -7,15 +7,15 @@ 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 PipelockProxy
 from tests.fixtures import fixture_minimal, fixture_with_ssh
 
 
 class TestPipelockProxyPrepare(unittest.TestCase):
     def setUp(self):
         self.out_dir = Path(tempfile.mkdtemp())
-        self.proxy = PipelockProxy()
+        self.proxy = DockerPipelockProxy()
 
     def tearDown(self):
         import shutil