Merge pull request 'refactor(cleanup): compose-ls driven + drop pipelock CIDR allowlist' (#36) from chunk-4-cleanup-cli into main

claude_bottle/backend/docker/bottle_cleanup_plan.py

@@ -1,8 +1,21 @@
 """DockerBottleCleanupPlan — concrete subclass of BottleCleanupPlan.
 
-Holds the tuples of container and network names that
+PRD 0018 chunk 4: cleanup is centered on compose projects. `docker
-DockerBottleBackend.cleanup will remove. The y/N preflight reads
+compose ls` is the source of truth for what's running; the plan
-these via `print`; the CLI short-circuits via `empty`.
+carries the projects to `compose down`, plus three fallback buckets
+for legacy / orphan resources:
+
+  - stray_containers: pre-compose `claude-bottle-*` containers not
+    attached to any compose project. Cleared via `docker rm -f`.
+  - stray_networks: same idea for networks. Cleared via
+    `docker network rm`.
+  - orphan_state_dirs: per-bottle state dirs under
+    ~/.claude-bottle/state/ that have no live compose project AND
+    no `.preserve` marker. Reaped via `shutil.rmtree`.
+
+Compose-managed networks are removed by `compose down --volumes`,
+so they don't appear in stray_networks for a normal project — only
+truly leftover ones.
 """
 
 from __future__ import annotations
@@ -17,20 +30,30 @@ 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
+    `prepare_cleanup`; sorted so the y/N output is stable."""
-    network ls`; sorted so the y/N output is stable."""
 
-    containers: tuple[str, ...]
+    projects: tuple[str, ...]
-    networks: tuple[str, ...]
+    stray_containers: tuple[str, ...]
+    stray_networks: tuple[str, ...]
+    orphan_state_dirs: tuple[str, ...]
 
     @property
     def empty(self) -> bool:
-        return not self.containers and not self.networks
+        return (
+            not self.projects
+            and not self.stray_containers
+            and not self.stray_networks
+            and not self.orphan_state_dirs
+        )
 
     def print(self) -> None:
         print(file=sys.stderr)
-        for name in self.containers:
+        for name in self.projects:
-            info(f"container: {name}")
+            info(f"compose project: {name}")
-        for name in self.networks:
+        for name in self.stray_containers:
-            info(f"network:   {name}")
+            info(f"stray container: {name}")
+        for name in self.stray_networks:
+            info(f"stray network:   {name}")
+        for name in self.orphan_state_dirs:
+            info(f"orphan state:    {name}")
         print(file=sys.stderr)

claude_bottle/backend/docker/cleanup.py

@@ -1,76 +1,180 @@
 """Cleanup + active-listing for the Docker bottle backend.
 
-`prepare_cleanup` enumerates orphaned `claude-bottle-` containers and
+PRD 0018 chunk 4: cleanup is centered on `docker compose ls`.
-networks; `cleanup` removes them. `list_active` queries the same
+Pre-compose code paths could leave bare containers / networks
-namespace for ad-hoc inspection. All three share a single concern:
+without a compose project; those still show up via the prefix
-acting on resources whose names start with `claude-bottle-`.
+scan, just as a fallback bucket alongside the project list.
+
+`prepare_cleanup` enumerates:
+
+  - Live compose projects whose name starts with `claude-bottle-`.
+  - `claude-bottle-*` containers that aren't part of any compose
+    project (legacy orphans).
+  - `claude-bottle-*` networks that aren't tied to a compose
+    project (legacy orphans; compose-managed networks come down
+    with `compose down --volumes` and don't appear here).
+  - State dirs under ~/.claude-bottle/state/<identity>/ with no
+    live compose project AND no `.preserve` marker.
+
+`cleanup` removes everything in the plan.
+
+`list_active` queries the same compose project namespace and prints
+each project's services for ad-hoc inspection.
 """
 
 from __future__ import annotations
 
+import json
+import shutil
 import subprocess
+from pathlib import Path
 
-from ...log import info
+from ... import supervise as _supervise
+from ...log import info, warn
 from . import util as docker_mod
 from .bottle_cleanup_plan import DockerBottleCleanupPlan
+from .bottle_state import bottle_state_dir, is_preserved
+
+
+_PROJECT_PREFIX = "claude-bottle-"
+
+
+def _list_compose_projects() -> list[str]:
+    """Return the names of all currently-known compose projects
+    (running OR stopped) whose name starts with `claude-bottle-`.
+    `docker compose ls --all` reports both up + exited states."""
+    result = subprocess.run(
+        ["docker", "compose", "ls", "--all", "--format", "json"],
+        capture_output=True, text=True, check=False,
+    )
+    if result.returncode != 0:
+        warn(f"docker compose ls failed: {result.stderr.strip()}")
+        return []
+    try:
+        projects = json.loads(result.stdout or "[]")
+    except json.JSONDecodeError as e:
+        warn(f"docker compose ls returned malformed JSON: {e}")
+        return []
+    names: list[str] = []
+    for p in projects:
+        if not isinstance(p, dict):
+            continue
+        name = str(p.get("Name", ""))
+        if name.startswith(_PROJECT_PREFIX):
+            names.append(name)
+    return sorted(set(names))
+
+
+def _list_prefixed_containers() -> list[str]:
+    """All claude-bottle-prefixed containers, running or stopped."""
+    result = subprocess.run(
+        ["docker", "ps", "-a",
+         "--filter", f"name=^{_PROJECT_PREFIX}",
+         "--format", "{{.Names}}\t{{.Label \"com.docker.compose.project\"}}"],
+        capture_output=True, text=True, check=False,
+    )
+    if result.returncode != 0:
+        warn(f"docker ps failed: {result.stderr.strip()}")
+        return []
+    out: list[str] = []
+    for line in (result.stdout or "").splitlines():
+        if not line:
+            continue
+        name, _, project = line.partition("\t")
+        # Stray = no compose label. Compose-managed containers carry
+        # `com.docker.compose.project=<name>`; we'll reap those via
+        # `compose down`, not via container rm.
+        if not project:
+            out.append(name)
+    return sorted(set(out))
+
+
+def _list_prefixed_networks() -> list[str]:
+    """All claude-bottle-prefixed networks not currently attached
+    to a compose project. Compose-managed networks have a
+    `com.docker.compose.project` label; bare ones (from pre-compose
+    code paths) don't."""
+    result = subprocess.run(
+        ["docker", "network", "ls",
+         "--filter", f"name={_PROJECT_PREFIX}",
+         "--format", "{{.Name}}\t{{.Label \"com.docker.compose.project\"}}"],
+        capture_output=True, text=True, check=False,
+    )
+    if result.returncode != 0:
+        warn(f"docker network ls failed: {result.stderr.strip()}")
+        return []
+    out: list[str] = []
+    for line in (result.stdout or "").splitlines():
+        if not line:
+            continue
+        name, _, project = line.partition("\t")
+        if not project:
+            out.append(name)
+    return sorted(set(out))
+
+
+def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
+    """State identities whose compose project isn't running and
+    that don't have a `.preserve` marker. `.preserve` means the
+    user (or an auto-preserve-on-crash) wants the state kept for
+    `resume`."""
+    state_root = _supervise.claude_bottle_root() / "state"
+    if not state_root.is_dir():
+        return []
+    orphans: list[str] = []
+    for child in sorted(state_root.iterdir()):
+        if not child.is_dir():
+            continue
+        identity = child.name
+        project = f"{_PROJECT_PREFIX}{identity}"
+        if project in live_projects:
+            continue
+        if is_preserved(identity):
+            continue
+        orphans.append(identity)
+    return orphans
 
 
 def prepare_cleanup() -> DockerBottleCleanupPlan:
-    """Enumerate all claude-bottle-prefixed containers (running or
+    """Enumerate everything cleanup will touch. No removals."""
-    stopped) and networks. No removals — caller confirms first."""
     docker_mod.require_docker()
-
+    projects = _list_compose_projects()
-    # `docker ps -a --filter name=...` uses regex matching; anchor at
+    project_set = set(projects)
-    # the start so we don't pick up containers that merely contain
+    return DockerBottleCleanupPlan(
-    # "claude-bottle-" mid-name.
+        projects=tuple(projects),
-    cr = subprocess.run(
+        stray_containers=tuple(_list_prefixed_containers()),
-        [
+        stray_networks=tuple(_list_prefixed_networks()),
-            "docker", "ps", "-a",
+        orphan_state_dirs=tuple(_list_orphan_state_dirs(project_set)),
-            "--filter", "name=^claude-bottle-",
-            "--format", "{{.Names}}",
-        ],
-        capture_output=True,
-        text=True,
-        check=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,
-        check=True,
-    )
-    networks = tuple(sorted(
-        line for line in (nr.stdout or "").splitlines() if line
-    ))
-
-    return DockerBottleCleanupPlan(containers=containers, networks=networks)
 
 
 def cleanup(plan: DockerBottleCleanupPlan) -> None:
-    """Remove the containers and networks listed in the plan.
+    """Remove everything in the plan. Projects first (whose `compose
-    Containers first; networks would refuse to delete while containers
+    down` reaps their containers + networks atomically), then stray
-    are still attached."""
+    legacy resources, then orphan state dirs."""
-    for name in plan.containers:
+    for project in plan.projects:
-        info(f"removing container {name}")
+        info(f"docker compose down  ({project})")
+        result = subprocess.run(
+            ["docker", "compose", "-p", project, "down", "--volumes"],
+            capture_output=True, text=True, check=False,
+        )
+        if result.returncode != 0:
+            warn(
+                f"compose down failed for {project}: "
+                f"{result.stderr.strip()}"
+            )
+
+    for name in plan.stray_containers:
+        info(f"removing stray container {name}")
         subprocess.run(
             ["docker", "rm", "-f", name],
             stdout=subprocess.DEVNULL,
             stderr=subprocess.DEVNULL,
             check=False,
         )
-    for name in plan.networks:
+
-        info(f"removing network {name}")
+    for name in plan.stray_networks:
+        info(f"removing stray network {name}")
         subprocess.run(
             ["docker", "network", "rm", name],
             stdout=subprocess.DEVNULL,
@@ -78,27 +182,50 @@ def cleanup(plan: DockerBottleCleanupPlan) -> None:
             check=False,
         )
 
+    for identity in plan.orphan_state_dirs:
+        path = bottle_state_dir(identity)
+        info(f"removing orphan state dir {path}")
+        try:
+            shutil.rmtree(path, ignore_errors=True)
+        except OSError as e:
+            warn(f"failed to remove {path}: {e}")
+
 
 def list_active() -> None:
-    """Print all running claude-bottle containers (name + status).
+    """Print every active claude-bottle compose project + its
-    Prints a single-line banner if there are none."""
+    services. Empty banner when there are none."""
     docker_mod.require_docker()
+    projects = _list_compose_projects()
+    # Filter to projects with at least one running container — `compose ls`
+    # already filters by default to active projects unless `--all` was
+    # set; double-check by querying status.
     result = subprocess.run(
-        [
+        ["docker", "compose", "ls", "--format", "json"],
-            "docker", "ps",
+        capture_output=True, text=True, check=False,
-            "--filter", "name=^claude-bottle-",
-            "--format", "{{.Names}}\t{{.Status}}",
-        ],
-        capture_output=True,
-        text=True,
-        check=True,
     )
-    containers = (result.stdout or "").strip()
+    running_names: set[str] = set()
-    if not containers:
+    if result.returncode == 0:
-        info("no active claude-bottle containers")
+        try:
+            data = json.loads(result.stdout or "[]")
+            running_names = {
+                str(p.get("Name", "")) for p in data if isinstance(p, dict)
+            }
+        except json.JSONDecodeError:
+            pass
+    active = [p for p in projects if p in running_names]
+    if not active:
+        info("no active claude-bottle compose projects")
         return
     print()
-    for line in containers.splitlines():
+    for project in active:
-        name, _, status = line.partition("\t")
+        info(f"compose project: {project}")
-        info(f"container: {name}  status: {status}")
+        ps = subprocess.run(
+            ["docker", "compose", "-p", project, "ps", "--format",
+             "{{.Service}}\t{{.Name}}\t{{.Status}}"],
+            capture_output=True, text=True, check=False,
+        )
+        for line in (ps.stdout or "").splitlines():
+            service, _, rest = line.partition("\t")
+            name, _, status = rest.partition("\t")
+            info(f"  {service:12s}  {name}  ({status})")
     print()

claude_bottle/backend/docker/compose.py

@@ -130,21 +130,18 @@ def bottle_plan_to_compose(plan: DockerBottlePlan) -> dict[str, Any]:
 
 
 def _networks(plan: DockerBottlePlan) -> dict[str, Any]:
-    """Both networks are `external: true` — chunk 3 pre-creates them
+    """Compose-managed networks with explicit `name:` matching the
-    via `docker network create` so pipelock's yaml can embed the
+    existing slug-suffixed convention. Compose creates them on `up`
-    internal-network CIDR in its SSRF allowlist before compose-up.
+    and destroys them on `down`. The internal one is `--internal`
-    Compose just references the pre-existing networks by name.
+    (no default gateway); the egress one is a normal user-defined
-    Network lifecycle (create / remove) is owned by the compose-
+    bridge."""
-    lifecycle helpers, not compose itself; `docker compose down`
-    leaves external networks alone."""
     return {
         "internal": {
             "name": plan.proxy_plan.internal_network,
-            "external": True,
+            "internal": True,
         },
         "egress": {
             "name": plan.proxy_plan.egress_network,
-            "external": True,
         },
     }
 

claude_bottle/backend/docker/launch.py

@@ -44,7 +44,6 @@ from typing import Callable, Generator
 
 from ...egress import egress_resolve_token_values
 from ...log import info
-from ...pipelock import pipelock_build_config, pipelock_render_yaml
 from . import network as network_mod
 from . import util as docker_mod
 from .bottle import DockerBottle
@@ -64,14 +63,9 @@ from .compose import (
     compose_up,
     write_compose_file,
 )
-from .egress import (
+from .egress import DockerEgress, egress_tls_init
-    DockerEgress,
-    egress_tls_init,
-)
 from .git_gate import DockerGitGate
 from .pipelock import (
-    PIPELOCK_CA_CERT_IN_CONTAINER,
-    PIPELOCK_CA_KEY_IN_CONTAINER,
     DockerPipelockProxy,
     pipelock_proxy_url,
     pipelock_tls_init,
@@ -124,44 +118,31 @@ def launch(
                 plan.derived_image, plan.image, plan.spec.user_cwd
             )
 
-        # Step 2: pre-create networks so we know the internal CIDR
+        # Networks: compose-managed. The names are derived
-        # before pipelock yaml renders.
+        # deterministically from the slug so the renderer can put
-        internal_network = network_mod.network_create_internal(plan.slug)
+        # them on the services and `compose up` creates them with
-        stack.callback(network_mod.network_remove, internal_network)
+        # those names. The empirical spike confirmed pipelock's
+        # SSRF guard only checks proxied-request destinations, not
+        # source IPs — so the bottle's own internal CIDR doesn't
+        # need to be in `ssrf.ip_allowlist`. Pre-create + CIDR
+        # introspection are gone; compose owns the network
+        # lifecycle.
+        internal_network = network_mod.network_name_for_slug(plan.slug)
+        egress_network = network_mod.network_egress_name_for_slug(plan.slug)
 
-        egress_network = network_mod.network_create_egress(plan.slug)
+        # Mint per-bottle CAs into state/<slug>/{pipelock,egress}/.
-        stack.callback(network_mod.network_remove, egress_network)
-
-        internal_cidr = network_mod.network_inspect_cidr(internal_network)
-
-        # Step 3: mint per-bottle CAs into state/<slug>/{pipelock,egress}/.
         ca_cert_host, ca_key_host = pipelock_tls_init(pipelock_state_dir(plan.slug))
         egress_ca_host, egress_ca_cert_only = egress_tls_init(
             egress_state_dir(plan.slug),
         )
 
-        # Step 4: re-render pipelock yaml with the SSRF allowlist now
+        # Populate launch-time fields on every inner plan so the
-        # that we know the internal CIDR. Prepare wrote the yaml
+        # renderer reads concrete network names, CA paths, and
-        # without the ssrf block; overwrite the same path so the
+        # pipelock URL.
-        # bind-mount picks up the updated content.
-        bottle = plan.spec.manifest.bottle_for(plan.spec.agent_name)
-        cfg = pipelock_build_config(
-            bottle,
-            ca_cert_path=PIPELOCK_CA_CERT_IN_CONTAINER,
-            ca_key_path=PIPELOCK_CA_KEY_IN_CONTAINER,
-            ssrf_ip_allowlist=(internal_cidr,),
-        )
-        plan.proxy_plan.yaml_path.write_text(pipelock_render_yaml(cfg))
-        plan.proxy_plan.yaml_path.chmod(0o600)
-
-        # Step 5: populate launch-time fields on every inner plan so
-        # the renderer reads concrete network names, CA paths, and
-        # pipelock URL. Match the field-by-field replacement the
-        # pre-compose launch did, just rolled into one pass.
         proxy_plan = dataclasses.replace(
             plan.proxy_plan,
             internal_network=internal_network,
-            internal_network_cidr=internal_cidr,
+            internal_network_cidr="",
             egress_network=egress_network,
             ca_cert_host_path=ca_cert_host,
             ca_key_host_path=ca_key_host,

claude_bottle/cli/cleanup.py

@@ -1,19 +1,22 @@
-"""cleanup: stop and remove all orphaned claude-bottle resources
+"""cleanup: stop and remove all orphaned claude-bottle resources.
-(containers + networks) left behind by previous bottles, plus
-optionally the per-bottle state dirs under ~/.claude-bottle/state/.
 
-State cleanup is prompted separately from container cleanup because
+PRD 0018 chunk 4: backend's prepare_cleanup carries everything in
-the trade-off is different: containers + networks are pure debris,
+one plan — live compose projects (whose `compose down` removes
-but a state dir may carry a resumable bottle (capability-block
+containers + networks atomically), legacy stray containers/networks
-rebuild + transcript snapshot) the operator still wants."""
+that aren't in any project, and orphan state dirs (per-bottle
+state with no live project AND no `.preserve` marker). One prompt,
+one cleanup call.
+
+State dirs with `.preserve` are intentionally never touched — they
+hold capability-block rebuilds or crash snapshots the operator may
+want to `resume`. Manual `rm -rf ~/.claude-bottle/state/<identity>`
+is the path for those.
+"""
 
 from __future__ import annotations
 
-import shutil
 import sys
-from pathlib import Path
 
-from .. import supervise as _supervise
 from ..backend import get_bottle_backend
 from ..log import info
 from ._common import read_tty_line
@@ -22,74 +25,21 @@ from ._common import read_tty_line
 def cmd_cleanup(_argv: list[str]) -> int:
     backend = get_bottle_backend()
     plan = backend.prepare_cleanup()
-    state_dirs = _enumerate_state_dirs()
 
-    if plan.empty and not state_dirs:
+    if plan.empty:
         info("no claude-bottle resources to clean up")
         return 0
 
-    if not plan.empty:
+    plan.print()
-        plan.print()
+    if not _prompt_yes("remove all of the above?"):
-        if _prompt_yes("remove all of the above?"):
+        info("cleanup: skipped")
-            backend.cleanup(plan)
+        return 0
-            info("containers + networks: cleaned")
-        else:
-            info("containers + networks: skipped")
-
-    if state_dirs:
-        _print_state(state_dirs)
-        if _prompt_yes(
-            "remove per-bottle state? (loses resumable bottles)",
-        ):
-            for d in state_dirs:
-                shutil.rmtree(d, ignore_errors=True)
-            info(f"state: removed {len(state_dirs)} dir(s)")
-        else:
-            info("state: skipped")
 
+    backend.cleanup(plan)
+    info("cleanup: done")
     return 0
 
 
-# --- State enumeration + display ------------------------------------------
-
-
-def _enumerate_state_dirs() -> list[Path]:
-    """All per-bottle state dirs under ~/.claude-bottle/state/.
-    Sorted for stable preflight output."""
-    state_root = _supervise.claude_bottle_root() / "state"
-    if not state_root.is_dir():
-        return []
-    return sorted(p for p in state_root.iterdir() if p.is_dir())
-
-
-def _state_summary(path: Path) -> str:
-    """One-line description suitable for the cleanup prompt. Calls
-    out resumability so the operator can decide whether removing it
-    loses anything they care about."""
-    flags: list[str] = []
-    if (path / "metadata.json").is_file():
-        flags.append("resumable")
-    else:
-        flags.append("no metadata.json (orphan)")
-    if (path / "Dockerfile").is_file():
-        flags.append("rebuilt Dockerfile")
-    if (path / "transcript").is_dir():
-        flags.append("transcript snapshot")
-    if (path / ".preserve").is_file():
-        flags.append("preserve marker")
-    return f"state:     {path.name} ({', '.join(flags)})"
-
-
-def _print_state(dirs: list[Path]) -> None:
-    print(file=sys.stderr)
-    for d in dirs:
-        info(_state_summary(d))
-    print(file=sys.stderr)
-
-
-# --- Prompt ----------------------------------------------------------------
-
-
 def _prompt_yes(message: str) -> bool:
     sys.stderr.write(f"claude-bottle: {message} [y/N] ")
     sys.stderr.flush()

tests/unit/test_cli_cleanup.py

@@ -1,102 +0,0 @@
-"""Unit: cli/cleanup.py state-dir enumeration + summary.
-
-The end-to-end cleanup-with-prompt flow is exercised manually;
-here we cover the state-dir display logic so a regression in the
-resumable / orphan / rebuild flags surfaces in unit CI."""
-
-import tempfile
-import unittest
-from pathlib import Path
-
-from claude_bottle import supervise
-from claude_bottle.backend.docker import bottle_state
-from claude_bottle.cli.cleanup import _enumerate_state_dirs, _state_summary
-
-
-class _FakeHomeMixin:
-    def _setup_fake_home(self):
-        self._tmp = tempfile.TemporaryDirectory(prefix="cli-cleanup-test.")
-        original = supervise.claude_bottle_root
-
-        def fake_root() -> Path:
-            return Path(self._tmp.name) / ".claude-bottle"
-
-        supervise.claude_bottle_root = fake_root  # type: ignore[assignment]
-        self._restore = lambda: setattr(supervise, "claude_bottle_root", original)
-
-    def _teardown_fake_home(self):
-        self._restore()
-        self._tmp.cleanup()
-
-
-class TestEnumerateStateDirs(_FakeHomeMixin, unittest.TestCase):
-    def setUp(self):
-        self._setup_fake_home()
-
-    def tearDown(self):
-        self._teardown_fake_home()
-
-    def test_empty_when_no_state_root(self):
-        self.assertEqual([], _enumerate_state_dirs())
-
-    def test_lists_each_identity_dir(self):
-        bottle_state.write_per_bottle_dockerfile("dev-aaa", "FROM x\n")
-        bottle_state.write_per_bottle_dockerfile("api-bbb", "FROM y\n")
-        dirs = _enumerate_state_dirs()
-        self.assertEqual(
-            ["api-bbb", "dev-aaa"],
-            [p.name for p in dirs],
-        )
-
-    def test_sorted_for_stable_preflight(self):
-        for name in ("z", "a", "m"):
-            bottle_state.write_per_bottle_dockerfile(name, "FROM x\n")
-        names = [p.name for p in _enumerate_state_dirs()]
-        self.assertEqual(["a", "m", "z"], names)
-
-
-class TestStateSummary(_FakeHomeMixin, unittest.TestCase):
-    def setUp(self):
-        self._setup_fake_home()
-
-    def tearDown(self):
-        self._teardown_fake_home()
-
-    def _path(self, name: str) -> Path:
-        return bottle_state.bottle_state_dir(name)
-
-    def test_orphan_state_dir(self):
-        # Only a Dockerfile, no metadata.json — the "api / dev" shape
-        # that comes from pre-identity-fix code.
-        bottle_state.write_per_bottle_dockerfile("orphan", "FROM old\n")
-        s = _state_summary(self._path("orphan"))
-        self.assertIn("orphan", s)
-        self.assertIn("no metadata.json", s)
-        self.assertIn("rebuilt Dockerfile", s)
-
-    def test_resumable_state_dir(self):
-        bottle_state.write_metadata(bottle_state.BottleMetadata(
-            identity="dev-aaa", agent_name="dev",
-            cwd="/proj/A", copy_cwd=True, started_at="t",
-        ))
-        s = _state_summary(self._path("dev-aaa"))
-        self.assertIn("resumable", s)
-        self.assertNotIn("rebuilt Dockerfile", s)
-
-    def test_resumable_with_capability_rebuild_and_preserve_marker(self):
-        bottle_state.write_metadata(bottle_state.BottleMetadata(
-            identity="dev-bbb", agent_name="dev",
-            cwd="", copy_cwd=False, started_at="t",
-        ))
-        bottle_state.write_per_bottle_dockerfile("dev-bbb", "FROM rebuilt\n")
-        bottle_state.transcript_snapshot_dir("dev-bbb").mkdir(parents=True)
-        bottle_state.mark_preserved("dev-bbb")
-        s = _state_summary(self._path("dev-bbb"))
-        self.assertIn("resumable", s)
-        self.assertIn("rebuilt Dockerfile", s)
-        self.assertIn("transcript snapshot", s)
-        self.assertIn("preserve marker", s)
-
-
-if __name__ == "__main__":
-    unittest.main()

tests/unit/test_compose.py

@@ -176,20 +176,19 @@ class TestProjectAndNetworks(unittest.TestCase):
         spec = bottle_plan_to_compose(_plan())
         self.assertEqual(f"claude-bottle-{SLUG}", spec["name"])
 
-    def test_internal_network_marked_external(self):
+    def test_internal_network_is_internal(self):
-        # Chunk 3 pre-creates networks with `docker network create
-        # --internal` so pipelock can know the CIDR before compose-up.
-        # Compose references the network by name with `external: true`.
         spec = bottle_plan_to_compose(_plan())
         net = spec["networks"]["internal"]
         self.assertEqual(f"claude-bottle-net-{SLUG}", net["name"])
-        self.assertTrue(net["external"])
+        self.assertTrue(net["internal"])
 
-    def test_egress_network_marked_external(self):
+    def test_egress_network_is_external_bridge(self):
         spec = bottle_plan_to_compose(_plan())
         net = spec["networks"]["egress"]
         self.assertEqual(f"claude-bottle-egress-{SLUG}", net["name"])
-        self.assertTrue(net["external"])
+        # No `internal:` key on the egress network — defaults to a
+        # normal user-defined bridge.
+        self.assertNotIn("internal", net)
 
 
 class TestPipelockAlwaysPresent(unittest.TestCase):

tests/unit/test_docker_cleanup.py

@@ -0,0 +1,102 @@
+"""Unit: backend/docker/cleanup orphan-state-dir detection.
+
+PRD 0018 chunk 4. The orphan-state-dir rule has three categories:
+  - LIVE: a compose project with the matching name is up → keep
+  - PRESERVED: state dir carries `.preserve` → keep (resume target)
+  - ORPHAN: neither → reap
+
+These are the cases the test exercises. The compose-project +
+container/network enumeration is left to the integration tests
+because it requires a real docker daemon."""
+
+from __future__ import annotations
+
+import tempfile
+import unittest
+from pathlib import Path
+
+from claude_bottle import supervise
+from claude_bottle.backend.docker import bottle_state
+from claude_bottle.backend.docker.cleanup import _list_orphan_state_dirs
+
+
+class _FakeHomeMixin:
+    def _setup_fake_home(self) -> None:
+        self._tmp = tempfile.TemporaryDirectory(prefix="docker-cleanup-test.")
+        original = supervise.claude_bottle_root
+
+        def fake_root() -> Path:
+            return Path(self._tmp.name) / ".claude-bottle"
+
+        supervise.claude_bottle_root = fake_root  # type: ignore[assignment]
+        self._restore = lambda: setattr(supervise, "claude_bottle_root", original)
+
+    def _teardown_fake_home(self) -> None:
+        self._restore()
+        self._tmp.cleanup()
+
+
+class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
+    def setUp(self) -> None:
+        self._setup_fake_home()
+
+    def tearDown(self) -> None:
+        self._teardown_fake_home()
+
+    def test_no_state_root_returns_empty(self):
+        self.assertEqual([], _list_orphan_state_dirs(set()))
+
+    def test_state_dir_with_no_live_no_preserve_is_orphan(self):
+        # Just touch the dir; no metadata, no preserve marker — the
+        # exact orphan shape.
+        bottle_state.write_per_bottle_dockerfile("solo-aaa", "FROM x\n")
+        self.assertEqual(
+            ["solo-aaa"],
+            _list_orphan_state_dirs(set()),
+        )
+
+    def test_live_project_skips_dir(self):
+        # Live project means the bottle is currently running under
+        # compose — never reap.
+        bottle_state.write_per_bottle_dockerfile("live-bbb", "FROM x\n")
+        self.assertEqual(
+            [],
+            _list_orphan_state_dirs({"claude-bottle-live-bbb"}),
+        )
+
+    def test_preserve_marker_skips_dir(self):
+        # Preserve marker = capability-block or crash auto-preserve;
+        # the user explicitly wanted this dir kept for `resume`.
+        bottle_state.write_per_bottle_dockerfile("kept-ccc", "FROM x\n")
+        bottle_state.mark_preserved("kept-ccc")
+        self.assertEqual(
+            [],
+            _list_orphan_state_dirs(set()),
+        )
+
+    def test_preserve_overrides_no_live_project(self):
+        # Even without a live project, a preserve marker keeps it.
+        bottle_state.write_per_bottle_dockerfile("kept-ddd", "FROM x\n")
+        bottle_state.mark_preserved("kept-ddd")
+        self.assertEqual([], _list_orphan_state_dirs(set()))
+
+    def test_mixed_set_categorized_correctly(self):
+        bottle_state.write_per_bottle_dockerfile("orphan-eee", "FROM x\n")
+        bottle_state.write_per_bottle_dockerfile("live-fff", "FROM y\n")
+        bottle_state.write_per_bottle_dockerfile("kept-ggg", "FROM z\n")
+        bottle_state.mark_preserved("kept-ggg")
+
+        result = _list_orphan_state_dirs({"claude-bottle-live-fff"})
+        self.assertEqual(["orphan-eee"], result)
+
+    def test_sorted_output(self):
+        for name in ("zzz-1", "aaa-1", "mmm-1"):
+            bottle_state.write_per_bottle_dockerfile(name, "FROM x\n")
+        self.assertEqual(
+            ["aaa-1", "mmm-1", "zzz-1"],
+            _list_orphan_state_dirs(set()),
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()