feat(cleanup): walk every backend, reap smolmachines orphans too

`./cli.py cleanup` previously called only the env-var-selected
backend's `prepare_cleanup` / `cleanup` — so a leftover smolvm
machine + bundle container + bundle network from a crashed
smolmachines bottle would survive a default `docker`-mode cleanup
indefinitely.

Smolmachines now has a real `cleanup` module (alongside
`enumerate.py` from issue #77) that walks:

  - smolvm machines named `claude-bottle-*` (via
    `smolvm machine ls --json`)
  - bundle containers `claude-bottle-sidecars-*`
  - bundle networks `claude-bottle-bundle-*`

Cleanup runs stop+delete on the machines, force-rm on the
containers, network rm on the networks. Each step is best-effort
so a failed rm doesn't block the rest.

`cli.py cleanup` walks every backend in `known_backend_names()`
and runs each backend's `cleanup` after a single y/N prompt that
shows a combined plan.

State dirs (`~/.claude-bottle/state/<slug>/`) are shared layout
with the docker backend, which still owns the orphan-state-dir
bucket. It now consults `enumerate_active_bottles()` for the
cross-backend live identity set so a running smolmachines
bottle's state dir isn't reaped during a cleanup.

Tests: smolmachines cleanup (prepare + cleanup ordering + failure
handling); cross-backend orphan protection on the docker
state-dir check; CLI cmd_cleanup walks both backends, short-
circuits on all-empty, aborts on N. 617 unit tests pass.

End-to-end verified on this host:
  $ smolvm machine ls --json | jq '.[].name'
  "claude-bottle-researcher-m3hxd"
  $ ./cli.py cleanup
  --- smolmachines backend ---
  smolvm machine:  claude-bottle-researcher-m3hxd
  remove all of the above? [y/N]

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>

claude_bottle/backend/__init__.py

@@ -139,6 +139,21 @@ class Bottle(ABC):
 
     name: str
 
+    @abstractmethod
+    def claude_argv(
+        self, argv: list[str], *, tty: bool = True,
+    ) -> list[str]:
+        """Return the host-side argv that runs `claude <argv>`
+        inside the bottle. Used by `exec_claude` for foreground
+        handoffs and by the dashboard's tmux `respawn-pane` flow,
+        which needs the argv up front (it spawns claude in a tmux
+        pane rather than as a child of the current process).
+
+        Implementations transparently inject
+        `--append-system-prompt-file` when the bottle was launched
+        with a provisioned prompt path."""
+        ...
+
     @abstractmethod
     def exec_claude(self, argv: list[str], *, tty: bool = True) -> int: ...
 

claude_bottle/backend/docker/bottle.py

@@ -28,15 +28,9 @@ class DockerBottle(Bottle):
         self._prompt_path = prompt_path_in_container
         self._closed = False
 
-    def claude_docker_argv(
+    def claude_argv(
         self, argv: list[str], *, tty: bool = True,
     ) -> list[str]:
-        """Return the full `docker exec` argv for running claude in
-        this bottle. Public so callers that want to spawn claude
-        somewhere other than the dashboard's foreground (e.g.,
-        `tmux split-window` / `tmux respawn-pane` from the dashboard
-        when `$TMUX` is set) can build on the same command without
-        duplicating the `--append-system-prompt-file` plumbing."""
         full_argv = list(argv)
         if self._prompt_path:
             full_argv.extend(["--append-system-prompt-file", self._prompt_path])
@@ -48,7 +42,7 @@ class DockerBottle(Bottle):
 
     def exec_claude(self, argv: list[str], *, tty: bool = True) -> int:
         return subprocess.run(
-            self.claude_docker_argv(argv, tty=tty), check=False,
+            self.claude_argv(argv, tty=tty), check=False,
         ).returncode
 
     def exec(self, script: str, *, user: str = "node") -> ExecResult:

claude_bottle/backend/docker/cleanup.py

@@ -83,11 +83,18 @@ def _list_prefixed_networks() -> list[str]:
     return sorted(set(out))
 
 
-def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
+def _list_orphan_state_dirs(
+    live_projects: set[str], protected_identities: 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`."""
+    `resume`.
+
+    `protected_identities` is the set of slugs that are live in
+    ANY backend — used so this docker-side check doesn't reap a
+    running smolmachines bottle's state dir (the layout is shared
+    across both backends)."""
     state_root = _supervise.claude_bottle_root() / "state"
     if not state_root.is_dir():
         return []
@@ -99,6 +106,8 @@ def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
         project = f"{COMPOSE_PROJECT_PREFIX}{identity}"
         if project in live_projects:
             continue
+        if identity in protected_identities:
+            continue
         if is_preserved(identity):
             continue
         orphans.append(identity)
@@ -106,15 +115,25 @@ def _list_orphan_state_dirs(live_projects: set[str]) -> list[str]:
 
 
 def prepare_cleanup() -> DockerBottleCleanupPlan:
-    """Enumerate everything cleanup will touch. No removals."""
+    """Enumerate everything cleanup will touch. No removals.
+
+    Pulls the union of live identities across backends via
+    `enumerate_active_agents()` so the orphan-state-dir bucket
+    doesn't include slugs whose smolmachines VM is still up."""
     docker_mod.require_docker()
     projects = list_compose_projects()
     project_set = set(projects)
+    # Late import to avoid a circular at module-load time —
+    # the backend package's __init__ imports this module.
+    from .. import enumerate_active_agents
+    protected = {a.slug for a in enumerate_active_agents()}
     return DockerBottleCleanupPlan(
         projects=tuple(projects),
         stray_containers=tuple(_list_prefixed_containers()),
         stray_networks=tuple(_list_prefixed_networks()),
-        orphan_state_dirs=tuple(_list_orphan_state_dirs(project_set)),
+        orphan_state_dirs=tuple(
+            _list_orphan_state_dirs(project_set, protected),
+        ),
     )
 
 

claude_bottle/backend/smolmachines/backend.py

@@ -8,6 +8,7 @@ from pathlib import Path
 from typing import Generator, Sequence
 
 from .. import ActiveAgent, BottleBackend, BottleSpec
+from . import cleanup as _cleanup
 from . import enumerate as _enumerate
 from . import launch as _launch
 from . import prepare as _prepare
@@ -76,12 +77,10 @@ class SmolmachinesBottleBackend(
         _supervise.provision_supervise(plan, target)
 
     def prepare_cleanup(self) -> SmolmachinesBottleCleanupPlan:
-        return SmolmachinesBottleCleanupPlan()
+        return _cleanup.prepare_cleanup()
 
     def cleanup(self, plan: SmolmachinesBottleCleanupPlan) -> None:
-        del plan
-        # Nothing to clean in chunks 1-3 — see
-        # SmolmachinesBottleCleanupPlan docstring.
+        _cleanup.cleanup(plan)
 
     def enumerate_active(self) -> Sequence[ActiveAgent]:
         return _enumerate.enumerate_active()

claude_bottle/backend/smolmachines/bottle.py

@@ -75,6 +75,21 @@ class SmolmachinesBottle(Bottle):
         # because exec doesn't inherit from machine_create's env.
         self._guest_env = dict(guest_env or {})
 
+    def claude_argv(
+        self, argv: list[str], *, tty: bool = True,
+    ) -> list[str]:
+        flags = ["smolvm", "machine", "exec", "--name", self.name]
+        if tty:
+            flags += ["-i", "-t"]
+        flags += _env_flags_for("node")
+        flags += _guest_env_flags(self._guest_env)
+        claude_tail = ["claude"]
+        if self._prompt_path:
+            claude_tail += ["--append-system-prompt-file", self._prompt_path]
+        claude_tail += argv
+        flags += ["--", "runuser", "-u", "node", "--", *claude_tail]
+        return flags
+
     def exec_claude(self, argv: list[str], *, tty: bool = True) -> int:
         """Run `claude` interactively inside the VM as the `node`
         user. Inherits the operator's terminal (stdin / stdout /
@@ -89,18 +104,9 @@ class SmolmachinesBottle(Bottle):
         UID switches via `runuser -u node --` (not `-l`) so we
         avoid login-shell wiring. HOME / USER come from `smolvm
         -e` instead, which sets them on the process env."""
-        flags = ["smolvm", "machine", "exec", "--name", self.name]
-        if tty:
-            flags += ["-i", "-t"]
-        flags += _env_flags_for("node")
-        flags += _guest_env_flags(self._guest_env)
-        claude_argv = ["claude"]
-        if self._prompt_path:
-            claude_argv += ["--append-system-prompt-file", self._prompt_path]
-        claude_argv += argv
-        flags += ["--", "runuser", "-u", "node", "--", *claude_argv]
-        result = subprocess.run(flags, check=False)
-        return result.returncode
+        return subprocess.run(
+            self.claude_argv(argv, tty=tty), check=False,
+        ).returncode
 
     def exec(self, script: str, *, user: str = "node") -> ExecResult:
         """Run a POSIX shell script as `user` (default `node`) and

claude_bottle/backend/smolmachines/bottle_cleanup_plan.py

@@ -1,13 +1,29 @@
-"""SmolmachinesBottleCleanupPlan — concrete BottleCleanupPlan stub
-(PRD 0023 chunk 1).
+"""SmolmachinesBottleCleanupPlan — concrete BottleCleanupPlan (issue #77).
 
-Chunk 1 always reports nothing-to-clean. Real enumeration —
-orphaned smolvm machines, stranded gvproxy sockets, leftover
-sidecar bundle containers — lands in chunk 4 alongside the
-integration-test sweep that exercises teardown."""
+Tracks the resources `SmolmachinesBottleBackend.cleanup` will
+remove:
+
+  - machines:  smolvm machines whose name starts with
+               `claude-bottle-` (running or stopped). Stopped +
+               deleted via `smolvm machine stop` + `machine delete -f`.
+  - bundles:   docker containers `claude-bottle-sidecars-<slug>`
+               left over from a smolmachines bottle (the bundle's
+               port-forwards stay published on lo0 aliases until
+               the container is gone). Removed via `docker rm -f`.
+  - networks:  docker networks `claude-bottle-bundle-<slug>`
+               attached to the bundles. Removed via
+               `docker network rm`.
+
+Smolmachines state dirs live under the same `~/.claude-bottle/state/`
+path the docker backend uses; the docker backend's
+`prepare_cleanup` already enumerates orphan state dirs and is the
+single source of truth for that bucket (consults
+`enumerate_active_bottles()` so it doesn't reap a live
+smolmachines bottle's dir)."""
 
 from __future__ import annotations
 
+import sys
 from dataclasses import dataclass
 
 from ...log import info
@@ -16,10 +32,24 @@ from .. import BottleCleanupPlan
 
 @dataclass(frozen=True)
 class SmolmachinesBottleCleanupPlan(BottleCleanupPlan):
-    def print(self) -> None:
-        info("smolmachines cleanup: nothing to remove (chunk 4 will "
-             "enumerate orphan machines + gvproxy sockets)")
+    """Resources SmolmachinesBottleBackend.cleanup will remove.
+    Produced by `prepare_cleanup`; sorted so the y/N output is
+    stable."""
+
+    machines: tuple[str, ...] = ()
+    bundles: tuple[str, ...] = ()
+    networks: tuple[str, ...] = ()
 
     @property
     def empty(self) -> bool:
-        return True
+        return not self.machines and not self.bundles and not self.networks
+
+    def print(self) -> None:
+        print(file=sys.stderr)
+        for name in self.machines:
+            info(f"smolvm machine:  {name}")
+        for name in self.bundles:
+            info(f"bundle container:{name}")
+        for name in self.networks:
+            info(f"bundle network:  {name}")
+        print(file=sys.stderr)

claude_bottle/backend/smolmachines/cleanup.py

@@ -0,0 +1,159 @@
+"""Cleanup + active-listing for the smolmachines backend (issue #77).
+
+`prepare_cleanup` enumerates leftover smolmachines resources:
+
+  - smolvm machines (`smolvm machine ls --json`) whose name starts
+    with `claude-bottle-`.
+  - bundle docker containers (`claude-bottle-sidecars-<slug>`).
+  - bundle docker networks (`claude-bottle-bundle-<slug>`).
+
+State dirs live under `~/.claude-bottle/state/<identity>/` —
+shared layout with the docker backend, which has the single
+orphan-state-dir enumerator (it already consults
+`enumerate_active_agents()` so a live smolmachines bottle's dir
+is preserved).
+
+`cleanup` removes everything in the plan: stop + delete each VM,
+force-rm each container, rm each network. Each step is
+best-effort — a failure on one resource doesn't block the others."""
+
+from __future__ import annotations
+
+import json
+import subprocess
+
+from ...log import info, warn
+from . import sidecar_bundle as _bundle
+from . import smolvm as _smolvm
+from .bottle_cleanup_plan import SmolmachinesBottleCleanupPlan
+
+
+# Both names start with the same prefix the launcher uses.
+_VM_PREFIX = "claude-bottle-"
+_BUNDLE_PREFIX = _bundle.bundle_container_name("")          # `claude-bottle-sidecars-`
+_NETWORK_PREFIX = _bundle.bundle_network_name("")           # `claude-bottle-bundle-`
+
+
+def prepare_cleanup() -> SmolmachinesBottleCleanupPlan:
+    """Enumerate every smolmachines-owned resource on the host.
+    No side effects. Returns an empty plan when smolvm isn't on
+    PATH (no machines to reap) — `cleanup` is a no-op in that
+    case too."""
+    machines = _list_claude_bottle_machines()
+    bundles = _list_bundle_containers()
+    networks = _list_bundle_networks()
+    return SmolmachinesBottleCleanupPlan(
+        machines=tuple(sorted(machines)),
+        bundles=tuple(sorted(bundles)),
+        networks=tuple(sorted(networks)),
+    )
+
+
+def cleanup(plan: SmolmachinesBottleCleanupPlan) -> None:
+    """Remove everything in the plan. Order matters: stop VMs
+    first (they hold ports on lo0 aliases via libkrun), then the
+    bundle containers (which hold the host port-forwards), then
+    the networks (which docker won't reap until the containers
+    are gone)."""
+    for name in plan.machines:
+        info(f"stopping smolvm machine {name}")
+        subprocess.run(
+            ["smolvm", "machine", "stop", "--name", name],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            check=False,
+        )
+        info(f"deleting smolvm machine {name}")
+        r = subprocess.run(
+            ["smolvm", "machine", "delete", "-f", name],
+            capture_output=True, text=True, check=False,
+        )
+        if r.returncode != 0:
+            warn(
+                f"smolvm machine delete -f {name} failed: "
+                f"{(r.stderr or '').strip()}"
+            )
+
+    for name in plan.bundles:
+        info(f"removing bundle container {name}")
+        subprocess.run(
+            ["docker", "rm", "-f", name],
+            stdout=subprocess.DEVNULL, stderr=subprocess.DEVNULL,
+            check=False,
+        )
+
+    for name in plan.networks:
+        info(f"removing bundle network {name}")
+        r = subprocess.run(
+            ["docker", "network", "rm", name],
+            capture_output=True, text=True, check=False,
+        )
+        if r.returncode != 0 and "no such network" not in (r.stderr or "").lower():
+            warn(
+                f"docker network rm {name} failed: "
+                f"{(r.stderr or '').strip()}"
+            )
+
+
+def _list_claude_bottle_machines() -> list[str]:
+    """All smolvm machines named `claude-bottle-*`, regardless of
+    state (running / stopped / created). Empty when smolvm isn't
+    installed."""
+    if not _smolvm.is_available():
+        return []
+    r = subprocess.run(
+        ["smolvm", "machine", "ls", "--json"],
+        capture_output=True, text=True, check=False,
+    )
+    if r.returncode != 0:
+        return []
+    try:
+        machines = json.loads(r.stdout or "[]")
+    except json.JSONDecodeError:
+        return []
+    return [
+        m["name"] for m in machines
+        if isinstance(m, dict)
+        and m.get("name", "").startswith(_VM_PREFIX)
+    ]
+
+
+def _list_bundle_containers() -> list[str]:
+    """All docker containers named `claude-bottle-sidecars-*`,
+    running or stopped. Empty when docker isn't installed."""
+    # Late import: `backend/__init__` imports this module
+    # transitively via the smolmachines backend.
+    from .. import has_backend
+    if not has_backend("docker"):
+        return []
+    r = subprocess.run(
+        ["docker", "ps", "-a",
+         "--filter", f"name=^{_BUNDLE_PREFIX}",
+         "--format", "{{.Names}}"],
+        capture_output=True, text=True, check=False,
+    )
+    if r.returncode != 0:
+        return []
+    return [
+        line for line in (r.stdout or "").splitlines()
+        if line and line.startswith(_BUNDLE_PREFIX)
+    ]
+
+
+def _list_bundle_networks() -> list[str]:
+    """All docker networks named `claude-bottle-bundle-*`. Empty
+    when docker isn't installed."""
+    from .. import has_backend
+    if not has_backend("docker"):
+        return []
+    r = subprocess.run(
+        ["docker", "network", "ls",
+         "--filter", f"name={_NETWORK_PREFIX}",
+         "--format", "{{.Name}}"],
+        capture_output=True, text=True, check=False,
+    )
+    if r.returncode != 0:
+        return []
+    return [
+        line for line in (r.stdout or "").splitlines()
+        if line and line.startswith(_NETWORK_PREFIX)
+    ]

claude_bottle/cli/cleanup.py

@@ -1,11 +1,16 @@
 """cleanup: stop and remove all orphaned claude-bottle resources.
 
-PRD 0018 chunk 4: backend's prepare_cleanup carries everything in
-one plan — live compose projects (whose `compose down` removes
-containers + networks atomically), legacy stray containers/networks
-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.
+Walks every registered backend (docker + smolmachines) so a single
+`./cli.py cleanup` reaps both backends' leftovers — orphaned
+smolvm machines won't survive a docker-only cleanup pass (issue
+addressed alongside #77).
+
+Each backend's `prepare_cleanup` enumerates its own resources;
+docker's `_list_orphan_state_dirs` consults
+`enumerate_active_agents()` for the union of live identities so
+state dirs of running smolmachines bottles aren't reaped. State
+dirs are shared layout, so docker is the single owner of that
+bucket.
 
 State dirs with `.preserve` are intentionally never touched — they
 hold capability-block rebuilds or crash snapshots the operator may
@@ -17,25 +22,37 @@ from __future__ import annotations
 
 import sys
 
-from ..backend import get_bottle_backend
+from ..backend import get_bottle_backend, known_backend_names
 from ..log import info
 from ._common import read_tty_line
 
 
 def cmd_cleanup(_argv: list[str]) -> int:
-    backend = get_bottle_backend()
-    plan = backend.prepare_cleanup()
+    # Order: stable backend iteration so the y/N output is
+    # deterministic across runs.
+    plans = [
+        (name, get_bottle_backend(name)) for name in known_backend_names()
+    ]
+    prepared = [(name, b, b.prepare_cleanup()) for name, b in plans]
 
-    if plan.empty:
+    if all(p.empty for _, _, p in prepared):
         info("no claude-bottle resources to clean up")
         return 0
 
-    plan.print()
+    for name, _, plan in prepared:
+        if plan.empty:
+            continue
+        info(f"--- {name} backend ---")
+        plan.print()
+
     if not _prompt_yes("remove all of the above?"):
         info("cleanup: skipped")
         return 0
 
-    backend.cleanup(plan)
+    for name, backend, plan in prepared:
+        if plan.empty:
+            continue
+        backend.cleanup(plan)
     info("cleanup: done")
     return 0
 

claude_bottle/cli/dashboard.py

@@ -762,7 +762,7 @@ def _in_tmux() -> bool:
 
 
 def _claude_runtime_args(*, resume: bool, remote_control: bool = False) -> list[str]:
-    """The argv the dashboard hands to `bottle.claude_docker_argv`
+    """The argv the dashboard hands to `bottle.claude_argv`
     on every attach — matches what `attach_claude` builds for the
     foreground handoff so both surfaces produce the same claude
     invocation."""
@@ -777,28 +777,35 @@ def _claude_runtime_args(*, resume: bool, remote_control: bool = False) -> list[
 def _build_resume_argv_with_fallback(
     bottle, *, remote_control: bool = False,
 ) -> list[str]:
-    """Build a docker-exec argv that runs `claude --continue` and
+    """Build a backend-exec argv that runs `claude --continue` and
     falls back to plain `claude` if no prior session exists.
 
     `--continue` exits non-zero when an agent has been spun up
     but never typed at — there's no transcript to resume. The
     shell-level `||` wrapper makes that case start a fresh
     session instead of crashing the pane. The trade-off: we
-    invoke `sh -c` inside the container, so the command is two
+    invoke `sh -c` inside the bottle, so the command is two
     `claude` invocations behind a tiny shell rather than one
     direct exec. Acceptable; the shell adds microseconds and
     the fallback only kicks in when --continue would have
-    failed anyway."""
+    failed anyway.
+
+    Works across backends because `bottle.claude_argv` always
+    surfaces the `claude` token preceded by the backend's exec
+    framing (docker: `docker exec -it <c>`; smolmachines:
+    `smolvm machine exec --name <m> -- runuser -u node --`).
+    Splitting at `claude` keeps the framing as the prefix and
+    wraps just the claude tail in `sh -c`."""
     base_args = ["--dangerously-skip-permissions"]
     if remote_control:
         base_args.append("--remote-control")
-    base_docker = bottle.claude_docker_argv(base_args)
-    # Split docker-prefix from the claude-and-args tail so we
-    # can compose `<claude…> --continue || <claude…>` inside
+    base_exec = bottle.claude_argv(base_args)
+    # Split exec-framing prefix from the claude-and-args tail so
+    # we can compose `<claude…> --continue || <claude…>` inside
     # `sh -c`. The `claude` token is the marker.
-    claude_idx = base_docker.index("claude")
-    prefix = base_docker[:claude_idx]
-    claude_cmd = " ".join(shlex.quote(a) for a in base_docker[claude_idx:])
+    claude_idx = base_exec.index("claude")
+    prefix = base_exec[:claude_idx]
+    claude_cmd = " ".join(shlex.quote(a) for a in base_exec[claude_idx:])
     return [
         *prefix,
         "sh", "-c",
@@ -806,23 +813,23 @@ def _build_resume_argv_with_fallback(
     ]
 
 
-def _build_split_pane_argv(docker_argv: list[str]) -> list[str]:
-    """Pure helper: wrap a docker-exec argv with `tmux split-window
+def _build_split_pane_argv(claude_argv: list[str]) -> list[str]:
+    """Pure helper: wrap a backend-exec argv with `tmux split-window
     -h -P -F '#{pane_id}'`. The `-P -F` combo tells tmux to print
     the new pane's id on stdout so we can track it for later
     `respawn-pane` calls."""
     return [
         "tmux", "split-window", "-h",
         "-P", "-F", "#{pane_id}",
-        *docker_argv,
+        *claude_argv,
     ]
 
 
-def _build_respawn_pane_argv(pane_id: str, docker_argv: list[str]) -> list[str]:
-    """Pure helper: wrap a docker-exec argv with `tmux respawn-pane
+def _build_respawn_pane_argv(pane_id: str, claude_argv: list[str]) -> list[str]:
+    """Pure helper: wrap a backend-exec argv with `tmux respawn-pane
     -k -t <pane_id>`. `-k` kills the existing process in the pane
     before respawning."""
-    return ["tmux", "respawn-pane", "-k", "-t", pane_id, *docker_argv]
+    return ["tmux", "respawn-pane", "-k", "-t", pane_id, *claude_argv]
 
 
 @contextlib.contextmanager
@@ -1046,12 +1053,12 @@ def _attach_in_tmux(
         # exists (agent spun up but never typed at). Wrap with a
         # shell-level fallback so the pane lands in a fresh
         # claude instead of crashing.
-        docker_argv = _build_resume_argv_with_fallback(bottle)
+        claude_argv = _build_resume_argv_with_fallback(bottle)
     else:
-        docker_argv = bottle.claude_docker_argv(
+        claude_argv = bottle.claude_argv(
             _claude_runtime_args(resume=False),
         )
-    pane_id = _ensure_right_pane(tmux_state, docker_argv)
+    pane_id = _ensure_right_pane(tmux_state, claude_argv)
     if pane_id is None:
         # tmux failed (missing binary, server died, size error).
         # One status-line failover to the curses handoff so the

tests/unit/test_cli_cleanup_cross_backend.py

@@ -0,0 +1,105 @@
+"""Unit: `cli.py cleanup` walks every backend (issue follow-up).
+
+Asserts cmd_cleanup queries each backend's `prepare_cleanup`,
+combines the y/N output, and runs each backend's `cleanup` when
+the operator confirms.  Mocks the backends and stdin."""
+
+from __future__ import annotations
+
+import sys
+import unittest
+from unittest.mock import patch, MagicMock
+
+from claude_bottle.cli import cleanup as cmd
+
+
+def _make_backend(empty: bool = True):
+    backend = MagicMock()
+    plan = MagicMock(empty=empty)
+    backend.prepare_cleanup.return_value = plan
+    backend.cleanup = MagicMock()
+    return backend, plan
+
+
+class TestCmdCleanup(unittest.TestCase):
+    def test_iterates_every_backend(self):
+        docker, docker_plan = _make_backend(empty=False)
+        smol, smol_plan = _make_backend(empty=False)
+        backends_by_name = {"docker": docker, "smolmachines": smol}
+
+        with patch.object(
+            cmd, "known_backend_names",
+            return_value=("docker", "smolmachines"),
+        ), patch.object(
+            cmd, "get_bottle_backend",
+            side_effect=lambda name: backends_by_name[name],
+        ), patch.object(
+            cmd, "_prompt_yes", return_value=True,
+        ):
+            self.assertEqual(0, cmd.cmd_cleanup([]))
+
+        docker.prepare_cleanup.assert_called_once()
+        smol.prepare_cleanup.assert_called_once()
+        docker.cleanup.assert_called_once_with(docker_plan)
+        smol.cleanup.assert_called_once_with(smol_plan)
+
+    def test_short_circuits_when_all_empty(self):
+        docker, _ = _make_backend(empty=True)
+        smol, _ = _make_backend(empty=True)
+        backends_by_name = {"docker": docker, "smolmachines": smol}
+
+        with patch.object(
+            cmd, "known_backend_names",
+            return_value=("docker", "smolmachines"),
+        ), patch.object(
+            cmd, "get_bottle_backend",
+            side_effect=lambda name: backends_by_name[name],
+        ), patch.object(
+            cmd, "_prompt_yes",
+        ) as prompt:
+            self.assertEqual(0, cmd.cmd_cleanup([]))
+        prompt.assert_not_called()
+        docker.cleanup.assert_not_called()
+        smol.cleanup.assert_not_called()
+
+    def test_abort_at_prompt_runs_nothing(self):
+        docker, _ = _make_backend(empty=False)
+        smol, _ = _make_backend(empty=True)
+        backends_by_name = {"docker": docker, "smolmachines": smol}
+
+        with patch.object(
+            cmd, "known_backend_names",
+            return_value=("docker", "smolmachines"),
+        ), patch.object(
+            cmd, "get_bottle_backend",
+            side_effect=lambda name: backends_by_name[name],
+        ), patch.object(
+            cmd, "_prompt_yes", return_value=False,
+        ):
+            self.assertEqual(0, cmd.cmd_cleanup([]))
+        docker.cleanup.assert_not_called()
+        smol.cleanup.assert_not_called()
+
+    def test_skips_empty_plans_when_others_have_work(self):
+        # docker has work, smolmachines doesn't — only docker.cleanup
+        # is called.
+        docker, docker_plan = _make_backend(empty=False)
+        smol, _ = _make_backend(empty=True)
+        backends_by_name = {"docker": docker, "smolmachines": smol}
+
+        with patch.object(
+            cmd, "known_backend_names",
+            return_value=("docker", "smolmachines"),
+        ), patch.object(
+            cmd, "get_bottle_backend",
+            side_effect=lambda name: backends_by_name[name],
+        ), patch.object(
+            cmd, "_prompt_yes", return_value=True,
+        ):
+            cmd.cmd_cleanup([])
+        docker.cleanup.assert_called_once_with(docker_plan)
+        smol.cleanup.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_dashboard_active_agents.py

@@ -369,7 +369,7 @@ class TestResumeArgvWithFallback(unittest.TestCase):
 
 
 class TestClaudeRuntimeArgs(unittest.TestCase):
-    """The argv passed to `bottle.claude_docker_argv` on each
+    """The argv passed to `bottle.claude_argv` on each
     attach. Locked here so the tmux + foreground paths build
     identical claude invocations."""
 

tests/unit/test_docker_bottle.py

@@ -1,6 +1,6 @@
 """Unit: DockerBottle's argv builder (PRD 0021 chunk 1).
 
-`claude_docker_argv` is the pure helper that `exec_claude` and the
+`claude_argv` is the pure helper that `exec_claude` and the
 PRD-0021 tmux helpers both build on. It encodes two non-trivial
 rules — the optional `--append-system-prompt-file` flag and the
 optional `-it` for TTY mode — that we lock down here so the tmux
@@ -22,16 +22,16 @@ def _bottle(prompt_path: str | None = None) -> DockerBottle:
     )
 
 
-class TestClaudeDockerArgv(unittest.TestCase):
+class TestClaudeArgv(unittest.TestCase):
     def test_minimal_argv_no_prompt(self):
-        argv = _bottle().claude_docker_argv([])
+        argv = _bottle().claude_argv([])
         self.assertEqual(
             ["docker", "exec", "-it", "claude-bottle-dev-abc", "claude"],
             argv,
         )
 
     def test_appends_passed_args_after_claude(self):
-        argv = _bottle().claude_docker_argv(
+        argv = _bottle().claude_argv(
             ["--dangerously-skip-permissions", "--continue"],
         )
         self.assertEqual(
@@ -41,7 +41,7 @@ class TestClaudeDockerArgv(unittest.TestCase):
         )
 
     def test_appends_prompt_file_flag_when_set(self):
-        argv = _bottle("/home/node/.claude-bottle-prompt.txt").claude_docker_argv(
+        argv = _bottle("/home/node/.claude-bottle-prompt.txt").claude_argv(
             ["--dangerously-skip-permissions"],
         )
         self.assertEqual(
@@ -53,30 +53,30 @@ class TestClaudeDockerArgv(unittest.TestCase):
         )
 
     def test_no_prompt_flag_when_none(self):
-        argv = _bottle(None).claude_docker_argv(["--continue"])
+        argv = _bottle(None).claude_argv(["--continue"])
         self.assertNotIn("--append-system-prompt-file", argv)
 
     def test_empty_prompt_string_is_treated_as_no_prompt(self):
         # Matches the existing exec_claude behavior: falsy
         # prompt_path means "skip the flag." The synth path in
         # dashboard.py relies on this when metadata is missing.
-        argv = _bottle("").claude_docker_argv(["--continue"])
+        argv = _bottle("").claude_argv(["--continue"])
         self.assertNotIn("--append-system-prompt-file", argv)
 
     def test_tty_false_drops_it_flag(self):
-        argv = _bottle().claude_docker_argv([], tty=False)
+        argv = _bottle().claude_argv([], tty=False)
         self.assertEqual(
             ["docker", "exec", "claude-bottle-dev-abc", "claude"],
             argv,
         )
 
     def test_caller_argv_not_mutated(self):
-        # `claude_docker_argv` builds `full_argv` from a copy, so a
+        # `claude_argv` builds `full_argv` from a copy, so a
         # caller passing a long-lived list (e.g., the dashboard's
         # _claude_args fixture) doesn't get extra flags appended to
         # it on subsequent calls.
         original = ["--continue"]
-        _bottle("/x").claude_docker_argv(original)
+        _bottle("/x").claude_argv(original)
         self.assertEqual(["--continue"], original)
 
 

tests/unit/test_docker_cleanup.py

@@ -44,7 +44,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         self._teardown_fake_home()
 
     def test_no_state_root_returns_empty(self):
-        self.assertEqual([], _list_orphan_state_dirs(set()))
+        self.assertEqual([], _list_orphan_state_dirs(set(), set()))
 
     def test_state_dir_with_no_live_no_preserve_is_orphan(self):
         # Just touch the dir; no metadata, no preserve marker — the
@@ -52,7 +52,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         bottle_state.write_per_bottle_dockerfile("solo-aaa", "FROM x\n")
         self.assertEqual(
             ["solo-aaa"],
-            _list_orphan_state_dirs(set()),
+            _list_orphan_state_dirs(set(), set()),
         )
 
     def test_live_project_skips_dir(self):
@@ -61,7 +61,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         bottle_state.write_per_bottle_dockerfile("live-bbb", "FROM x\n")
         self.assertEqual(
             [],
-            _list_orphan_state_dirs({"claude-bottle-live-bbb"}),
+            _list_orphan_state_dirs({"claude-bottle-live-bbb"}, set()),
         )
 
     def test_preserve_marker_skips_dir(self):
@@ -71,14 +71,14 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         bottle_state.mark_preserved("kept-ccc")
         self.assertEqual(
             [],
-            _list_orphan_state_dirs(set()),
+            _list_orphan_state_dirs(set(), 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()))
+        self.assertEqual([], _list_orphan_state_dirs(set(), set()))
 
     def test_mixed_set_categorized_correctly(self):
         bottle_state.write_per_bottle_dockerfile("orphan-eee", "FROM x\n")
@@ -86,7 +86,7 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
         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"})
+        result = _list_orphan_state_dirs({"claude-bottle-live-fff"}, set())
         self.assertEqual(["orphan-eee"], result)
 
     def test_sorted_output(self):
@@ -94,7 +94,31 @@ class TestOrphanStateDirs(_FakeHomeMixin, unittest.TestCase):
             bottle_state.write_per_bottle_dockerfile(name, "FROM x\n")
         self.assertEqual(
             ["aaa-1", "mmm-1", "zzz-1"],
-            _list_orphan_state_dirs(set()),
+            _list_orphan_state_dirs(set(), set()),
+        )
+
+    def test_protected_identity_skips_dir(self):
+        # `protected_identities` carries slugs that are live in
+        # any backend (smolmachines included). docker's orphan
+        # detection respects them so a running smolmachines
+        # bottle's state dir isn't reaped while the VM is up.
+        bottle_state.write_per_bottle_dockerfile("smol-hhh", "FROM x\n")
+        self.assertEqual(
+            [],
+            _list_orphan_state_dirs(set(), {"smol-hhh"}),
+        )
+
+    def test_protected_overrides_no_live_project(self):
+        # A smolmachines bottle has no docker compose project but
+        # IS in the protected set; the absence of a project
+        # shouldn't cause a reap.
+        bottle_state.write_per_bottle_dockerfile("smol-iii", "FROM x\n")
+        self.assertEqual(
+            [],
+            _list_orphan_state_dirs(
+                {"claude-bottle-something-else"},  # different project up
+                {"smol-iii"},                       # but smol-iii is live
+            ),
         )
 
 

tests/unit/test_smolmachines_bottle.py

@@ -0,0 +1,107 @@
+"""Unit: SmolmachinesBottle's `claude_argv` builder.
+
+The dashboard's tmux pane-respawn path calls `bottle.claude_argv`
+directly (it spawns claude inside a tmux pane rather than as a
+child of the current process), so the argv shape is the
+non-trivial part. `exec_claude` is a thin wrapper around the same
+builder + `subprocess.run`; we lock the shape here.
+"""
+
+from __future__ import annotations
+
+import unittest
+
+from claude_bottle.backend.smolmachines.bottle import SmolmachinesBottle
+
+
+def _bottle(prompt_path: str | None = None, **env: str) -> SmolmachinesBottle:
+    return SmolmachinesBottle(
+        "claude-bottle-dev-abc",
+        prompt_path=prompt_path,
+        guest_env=env,
+    )
+
+
+class TestClaudeArgv(unittest.TestCase):
+    def test_minimal_argv_no_prompt(self):
+        argv = _bottle().claude_argv([])
+        self.assertEqual(
+            [
+                "smolvm", "machine", "exec", "--name",
+                "claude-bottle-dev-abc",
+                "-i", "-t",
+                "-e", "HOME=/home/node",
+                "-e", "USER=node",
+                "--",
+                "runuser", "-u", "node", "--",
+                "claude",
+            ],
+            argv,
+        )
+
+    def test_appends_passed_args_after_claude(self):
+        argv = _bottle().claude_argv(
+            ["--dangerously-skip-permissions", "--continue"],
+        )
+        # The claude tail is at the end of the argv, after the
+        # `runuser -u node --` switch.
+        self.assertEqual(
+            ["claude", "--dangerously-skip-permissions", "--continue"],
+            argv[argv.index("claude"):],
+        )
+
+    def test_appends_prompt_file_flag_when_set(self):
+        argv = _bottle("/home/node/.claude-bottle-prompt.txt").claude_argv(
+            ["--dangerously-skip-permissions"],
+        )
+        self.assertEqual(
+            [
+                "claude",
+                "--append-system-prompt-file",
+                "/home/node/.claude-bottle-prompt.txt",
+                "--dangerously-skip-permissions",
+            ],
+            argv[argv.index("claude"):],
+        )
+
+    def test_no_prompt_flag_when_none(self):
+        argv = _bottle(None).claude_argv(["--continue"])
+        self.assertNotIn("--append-system-prompt-file", argv)
+
+    def test_empty_prompt_string_is_treated_as_no_prompt(self):
+        argv = _bottle("").claude_argv(["--continue"])
+        self.assertNotIn("--append-system-prompt-file", argv)
+
+    def test_tty_false_drops_it_flags(self):
+        argv = _bottle().claude_argv([], tty=False)
+        self.assertNotIn("-i", argv)
+        self.assertNotIn("-t", argv)
+
+    def test_guest_env_forwarded_as_e_flags(self):
+        argv = _bottle(
+            None,
+            HTTPS_PROXY="http://127.0.0.1:1234",
+            NO_PROXY="localhost",
+        ).claude_argv([])
+        # `-e K=V` pairs land before the `--`. Order isn't
+        # guaranteed across dict iterations on older Pythons, but
+        # both must appear.
+        self.assertIn("-e", argv)
+        self.assertIn("HTTPS_PROXY=http://127.0.0.1:1234", argv)
+        self.assertIn("NO_PROXY=localhost", argv)
+
+    def test_runuser_switch_precedes_claude(self):
+        # The dashboard's `_build_resume_argv_with_fallback` finds
+        # the `claude` token to split exec-framing from the claude
+        # tail. `runuser -u node --` must sit on the prefix side so
+        # the shell wrap inherits the UID switch.
+        argv = _bottle().claude_argv([])
+        claude_idx = argv.index("claude")
+        self.assertEqual(
+            ["runuser", "-u", "node", "--"],
+            argv[claude_idx - 4:claude_idx],
+        )
+
+
+if __name__ == "__main__":
+    unittest.main()

tests/unit/test_smolmachines_cleanup.py

@@ -0,0 +1,148 @@
+"""Unit: smolmachines backend cleanup (`cleanup.py` +
+`bottle_cleanup_plan.py`).
+
+Tests mock `subprocess.run` + `has_backend` so they execute
+without docker / smolvm on PATH. Each cleanup step verifies argv
+shape; teardown verifies order (machines → bundles → networks)."""
+
+from __future__ import annotations
+
+import subprocess
+import unittest
+from unittest.mock import patch
+
+from claude_bottle import backend as backend_mod
+from claude_bottle.backend.smolmachines import cleanup
+from claude_bottle.backend.smolmachines.bottle_cleanup_plan import (
+    SmolmachinesBottleCleanupPlan,
+)
+
+
+def _ok(stdout: str = "", stderr: str = "") -> subprocess.CompletedProcess:
+    return subprocess.CompletedProcess(
+        args=[], returncode=0, stdout=stdout, stderr=stderr,
+    )
+
+
+class TestPrepareCleanup(unittest.TestCase):
+    def test_empty_when_nothing_running(self):
+        with patch.object(cleanup, "_smolvm") as smolvm, \
+             patch.object(cleanup.subprocess, "run") as run, \
+             patch.object(backend_mod, "has_backend", return_value=True):
+            smolvm.is_available.return_value = True
+            run.return_value = _ok(stdout="[]")
+            plan = cleanup.prepare_cleanup()
+        self.assertTrue(plan.empty)
+
+    def test_lists_machines_bundles_networks(self):
+        def fake_run(argv, *a, **kw):
+            if argv[:3] == ["smolvm", "machine", "ls"]:
+                return _ok(stdout=(
+                    '[{"name":"claude-bottle-a-1","state":"running"},'
+                    ' {"name":"claude-bottle-b-2","state":"created"},'
+                    ' {"name":"unrelated","state":"running"}]'
+                ))
+            if argv[:2] == ["docker", "ps"]:
+                return _ok(stdout=(
+                    "claude-bottle-sidecars-a-1\n"
+                    "claude-bottle-sidecars-b-2\n"
+                ))
+            if argv[:3] == ["docker", "network", "ls"]:
+                return _ok(stdout=(
+                    "claude-bottle-bundle-a-1\n"
+                    "claude-bottle-bundle-b-2\n"
+                ))
+            return _ok()
+
+        with patch.object(cleanup, "_smolvm") as smolvm, \
+             patch.object(cleanup.subprocess, "run", side_effect=fake_run), \
+             patch.object(backend_mod, "has_backend", return_value=True):
+            smolvm.is_available.return_value = True
+            plan = cleanup.prepare_cleanup()
+
+        # `unrelated` filtered out (no claude-bottle- prefix).
+        self.assertEqual(
+            ("claude-bottle-a-1", "claude-bottle-b-2"),
+            plan.machines,
+        )
+        self.assertEqual(
+            ("claude-bottle-sidecars-a-1", "claude-bottle-sidecars-b-2"),
+            plan.bundles,
+        )
+        self.assertEqual(
+            ("claude-bottle-bundle-a-1", "claude-bottle-bundle-b-2"),
+            plan.networks,
+        )
+
+    def test_no_smolvm_means_no_machines(self):
+        with patch.object(cleanup, "_smolvm") as smolvm, \
+             patch.object(cleanup.subprocess, "run", return_value=_ok()), \
+             patch.object(backend_mod, "has_backend", return_value=True):
+            smolvm.is_available.return_value = False
+            plan = cleanup.prepare_cleanup()
+        self.assertEqual((), plan.machines)
+
+
+class TestCleanup(unittest.TestCase):
+    def test_machines_stopped_then_deleted_then_bundles_then_networks(self):
+        plan = SmolmachinesBottleCleanupPlan(
+            machines=("claude-bottle-a-1",),
+            bundles=("claude-bottle-sidecars-a-1",),
+            networks=("claude-bottle-bundle-a-1",),
+        )
+        calls: list[list[str]] = []
+
+        def fake_run(argv, *a, **kw):
+            calls.append(list(argv[:4]))
+            return _ok()
+
+        with patch.object(cleanup.subprocess, "run", side_effect=fake_run):
+            cleanup.cleanup(plan)
+
+        # Stop precedes delete precedes bundle rm precedes network rm.
+        self.assertEqual(
+            ["smolvm", "machine", "stop", "--name"], calls[0],
+        )
+        self.assertEqual(
+            ["smolvm", "machine", "delete", "-f"], calls[1],
+        )
+        self.assertEqual(
+            ["docker", "rm", "-f", "claude-bottle-sidecars-a-1"], calls[2],
+        )
+        self.assertEqual(
+            ["docker", "network", "rm", "claude-bottle-bundle-a-1"], calls[3],
+        )
+
+    def test_failures_are_warnings_not_fatal(self):
+        # smolvm machine delete -f returning non-zero should warn
+        # but continue with bundles + networks. The cleanup is
+        # idempotent on success and tries to remove every resource.
+        plan = SmolmachinesBottleCleanupPlan(
+            machines=("claude-bottle-a-1",),
+            bundles=("claude-bottle-sidecars-a-1",),
+            networks=(),
+        )
+        results = iter([
+            _ok(),  # stop succeeds
+            subprocess.CompletedProcess(args=[], returncode=1, stdout="", stderr="boom"),  # delete fails
+            _ok(),  # bundle rm succeeds
+        ])
+
+        def fake_run(argv, *a, **kw):
+            return next(results)
+
+        with patch.object(cleanup.subprocess, "run", side_effect=fake_run), \
+             patch.object(cleanup, "warn") as warn:
+            cleanup.cleanup(plan)
+        # warn called once for the delete failure.
+        warn.assert_called_once()
+
+    def test_empty_plan_is_noop(self):
+        plan = SmolmachinesBottleCleanupPlan()
+        with patch.object(cleanup.subprocess, "run") as run:
+            cleanup.cleanup(plan)
+        run.assert_not_called()
+
+
+if __name__ == "__main__":
+    unittest.main()