feat(cli): cross-backend list active + --backend flag + dashboard picker (issue #77)
test / unit (pull_request) Successful in 29s
test / integration (pull_request) Successful in 41s

CLI and dashboard now share one cross-backend abstraction for
listing + launching bottles, so adding a backend (docker /
smolmachines) lights up in both places without separate wiring.

Backend abstraction:
- New `ActiveBottle` dataclass (`backend_name`, `slug`,
  `agent_name`, `started_at`, `services`) replaces the
  docker-specific `ActiveAgent`. Same field surface for the
  existing dashboard consumers; `ActiveAgent` becomes a typed
  alias for source-compat.
- New `BottleBackend.enumerate_active() -> Sequence[ActiveBottle]`
  replaces the old `list_active() -> None` (which printed and
  returned nothing). Docker implements it via the existing
  compose query; smolmachines implements it via `smolvm machine
  ls --json` cross-referenced with each bundle container's
  `CLAUDE_BOTTLE_SIDECAR_DAEMONS` env (`backend/smolmachines/
  enumerate.py`).
- New `enumerate_active_bottles()` and `known_backend_names()`
  module-level helpers fold every backend into one call.
- `get_bottle_backend(name=None)` takes an optional explicit
  name (precedence: arg > $CLAUDE_BOTTLE_BACKEND > "docker").

CLI:
- `./cli.py list active` enumerates every backend, prints
  tab-separated `<backend>\t<slug>\t<agent>\t<services>`. The
  smolmachines bottle the user was looking for now shows up.
- `./cli.py start` grows `--backend=<docker|smolmachines>`
  (choices pulled live from `known_backend_names()`). Threaded
  through `prepare_with_preflight(backend_name=...)` so the
  resume path picks up the flag too.

Dashboard:
- Active agents pane lists both backends (the row formatter now
  prefixes `[docker]` / `[smolmachines]`).
- New-agent flow inserts a backend picker modal between agent
  pick and preflight (`_backend_picker_modal`). Short-circuits
  when only one backend is registered.
- `discover_active_agents()` collapses to
  `enumerate_active_bottles()`; `_parse_services_by_project` and
  `_query_services_by_project` move to
  `backend/docker/cleanup.py` where the docker enumerator owns
  them.

Tests: parser + enumerate-active tests relocated to
`test_docker_enumerate_active.py`. New
`test_backend_selection.py` covers `get_bottle_backend`,
`known_backend_names`, `enumerate_active_bottles`. New
`test_cli_start_backend_flag.py` covers `--backend`'s argparse
shape + the explicit-over-env precedence.

605 unit tests pass.

Co-Authored-By: Claude Opus 4.7 <noreply@anthropic.com>
This commit is contained in:
2026-05-27 18:27:12 -04:00
parent 1e82aed54b
commit adff1263d8
12 changed files with 761 additions and 282 deletions
+69 -16
View File
@@ -19,12 +19,14 @@ backend exposes five methods:
cleanup(plan) -> None
Actually removes everything described by the cleanup plan.
list_active() -> None
Print every currently-running bottle on this backend to stderr.
enumerate_active() -> Sequence[ActiveBottle]
Return every currently-running bottle on this backend, with
enough metadata for callers (CLI `list active`, dashboard
agents pane) to render a row.
Selection is driven by CLAUDE_BOTTLE_BACKEND (default "docker"). Per
PRD 0003 the manifest does not carry a backend field; the host
environment picks.
Selection is driven by `--backend` on `start` or
CLAUDE_BOTTLE_BACKEND (env var; default "docker"). Per PRD 0003 the
manifest does not carry a backend field; the host picks.
"""
from __future__ import annotations
@@ -103,6 +105,26 @@ class ExecResult:
stderr: str
@dataclass(frozen=True)
class ActiveBottle:
"""One currently-running bottle, as the CLI `list active` and
dashboard agents pane render it.
Fields are deliberately backend-neutral. `services` is the set
of sidecar daemons currently up for this bottle (`pipelock`,
`egress`, `git-gate`, `supervise`); the dashboard uses it to
gate edit verbs. `backend_name` is the matching key in
`_BACKENDS` (`docker` / `smolmachines`) — used by the active-
list rendering to disambiguate and by the dashboard's
re-attach path."""
backend_name: str
slug: str
agent_name: str # from metadata.json; "?" if missing
started_at: str # ISO 8601 from metadata.json; "" if missing
services: tuple[str, ...] # alphabetical
class Bottle(ABC):
"""Handle to a running bottle. Yielded by a backend's launch step.
@@ -280,9 +302,11 @@ class BottleBackend(ABC, Generic[PlanT, CleanupT]):
"""Remove everything described by the cleanup plan."""
@abstractmethod
def list_active(self) -> None:
"""Print every currently-running bottle on this backend to
stderr (name + status)."""
def enumerate_active(self) -> Sequence[ActiveBottle]:
"""Return every currently-running bottle on this backend.
Empty when none. Backend-specific: docker queries `docker
compose ls`; smolmachines queries `smolvm machine ls --json`
+ cross-references its bundle container."""
# Import concrete backend classes AFTER the base types are defined, so
@@ -302,23 +326,52 @@ _BACKENDS: dict[str, BottleBackend[Any, Any]] = {
}
def get_bottle_backend() -> BottleBackend[Any, Any]:
"""Resolve the bottle backend for the active environment. Dies with
a pointer at the known backends if CLAUDE_BOTTLE_BACKEND names an
unimplemented one."""
name = os.environ.get("CLAUDE_BOTTLE_BACKEND", "docker")
if name not in _BACKENDS:
def get_bottle_backend(
name: str | None = None,
) -> BottleBackend[Any, Any]:
"""Resolve the bottle backend.
`name` precedence:
1. explicit arg (CLI `--backend=<name>` passes through here)
2. CLAUDE_BOTTLE_BACKEND env var
3. default `docker`
Dies with a pointer at the known backends if the chosen name
isn't implemented."""
resolved = name or os.environ.get("CLAUDE_BOTTLE_BACKEND") or "docker"
if resolved not in _BACKENDS:
known = ", ".join(sorted(_BACKENDS))
die(f"unknown CLAUDE_BOTTLE_BACKEND={name!r}; known backends: {known}")
return _BACKENDS[name]
die(f"unknown backend {resolved!r}; known backends: {known}")
return _BACKENDS[resolved]
def known_backend_names() -> tuple[str, ...]:
"""Sorted tuple of all backend keys in `_BACKENDS`. Used by
argparse (`--backend` choices) and the dashboard's backend
picker."""
return tuple(sorted(_BACKENDS))
def enumerate_active_bottles() -> list[ActiveBottle]:
"""All currently-running bottles, across every backend. Used by
CLI `list active` and the dashboard's agents pane so neither
has to know which backends exist. Ordered by backend name,
then slug."""
out: list[ActiveBottle] = []
for name in known_backend_names():
out.extend(_BACKENDS[name].enumerate_active())
return out
__all__ = [
"ActiveBottle",
"Bottle",
"BottleBackend",
"BottleCleanupPlan",
"BottlePlan",
"BottleSpec",
"ExecResult",
"enumerate_active_bottles",
"get_bottle_backend",
"known_backend_names",
]
+4 -4
View File
@@ -14,9 +14,9 @@ from __future__ import annotations
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from typing import Generator, Sequence
from .. import BottleBackend, BottleSpec
from .. import ActiveBottle, BottleBackend, BottleSpec
from . import cleanup as _cleanup
from . import launch as _launch
from . import prepare as _prepare
@@ -65,5 +65,5 @@ class DockerBottleBackend(BottleBackend["DockerBottlePlan", "DockerBottleCleanup
def cleanup(self, plan: DockerBottleCleanupPlan) -> None:
_cleanup.cleanup(plan)
def list_active(self) -> None:
_cleanup.list_active()
def enumerate_active(self) -> Sequence[ActiveBottle]:
return _cleanup.enumerate_active()
+70 -21
View File
@@ -29,10 +29,16 @@ import subprocess
from ... import supervise as _supervise
from ...log import info, warn
from .. import ActiveBottle
from . import util as docker_mod
from .bottle_cleanup_plan import DockerBottleCleanupPlan
from .bottle_state import bottle_state_dir, is_preserved
from .compose import COMPOSE_PROJECT_PREFIX, list_compose_projects
from .bottle_state import bottle_state_dir, is_preserved, read_metadata
from .compose import (
COMPOSE_PROJECT_PREFIX,
compose_project_name,
list_active_slugs,
list_compose_projects,
)
def _list_prefixed_containers() -> list[str]:
@@ -161,24 +167,67 @@ def cleanup(plan: DockerBottleCleanupPlan) -> None:
warn(f"failed to remove {path}: {e}")
def list_active() -> None:
"""Print every active claude-bottle compose project + its
services. Empty banner when there are none."""
docker_mod.require_docker()
active = list_compose_projects(include_stopped=False)
if not active:
info("no active claude-bottle compose projects")
return
print()
for project in active:
info(f"compose project: {project}")
ps = subprocess.run(
["docker", "compose", "-p", project, "ps", "--format",
"{{.Service}}\t{{.Name}}\t{{.Status}}"],
def enumerate_active() -> list[ActiveBottle]:
"""All currently-running docker-backed bottles as
`ActiveBottle` records. Backend-agnostic shape — the CLI
`list active` command and the dashboard agents pane both
consume this. Empty list when docker is unreachable or
nothing's running."""
# docker on PATH? Defensive — `list active` shouldn't die
# just because the docker backend isn't usable on this host.
if shutil.which("docker") is None:
return []
slugs = list_active_slugs(include_stopped=False)
if not slugs:
return []
services_by_project = _query_services_by_project()
out: list[ActiveBottle] = []
for slug in slugs:
project = compose_project_name(slug)
services = services_by_project.get(project, set())
metadata = read_metadata(slug)
out.append(ActiveBottle(
backend_name="docker",
slug=slug,
agent_name=metadata.agent_name if metadata else "?",
started_at=metadata.started_at if metadata else "",
services=tuple(sorted(services)),
))
return out
def _parse_services_by_project(stdout: str) -> dict[str, set[str]]:
"""Parse `docker ps` output formatted as
`<project-label>\\t<service-label>` (one line per container)
into a `{project: {service, ...}}` mapping. Pure function for
testing — the docker invocation is in `_query_services_by_project`."""
out: dict[str, set[str]] = {}
for line in stdout.splitlines():
project, _, service = line.partition("\t")
if not project or not service:
continue
out.setdefault(project, set()).add(service)
return out
def _query_services_by_project() -> dict[str, set[str]]:
"""One `docker ps` call → `{project: {service, ...}}`. Moved
here from the dashboard so the same query backs the CLI's
`list active` and the dashboard's agents pane."""
try:
r = subprocess.run(
[
"docker", "ps",
"--filter", "label=com.docker.compose.project",
"--format",
'{{.Label "com.docker.compose.project"}}'
"\t"
'{{.Label "com.docker.compose.service"}}',
],
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()
except FileNotFoundError:
return {}
if r.returncode != 0:
return {}
return _parse_services_by_project(r.stdout or "")
@@ -5,9 +5,10 @@ from __future__ import annotations
from contextlib import contextmanager
from pathlib import Path
from typing import Generator
from typing import Generator, Sequence
from .. import BottleBackend, BottleSpec
from .. import ActiveBottle, BottleBackend, BottleSpec
from . import enumerate as _enumerate
from . import launch as _launch
from . import prepare as _prepare
from .bottle import SmolmachinesBottle
@@ -73,9 +74,5 @@ class SmolmachinesBottleBackend(
# Nothing to clean in chunks 1-3 — see
# SmolmachinesBottleCleanupPlan docstring.
def list_active(self) -> None:
from ...log import info
info(
"smolmachines list_active: not implemented (chunk 4 wires "
"it to `smolvm machine ls --json`)"
)
def enumerate_active(self) -> Sequence[ActiveBottle]:
return _enumerate.enumerate_active()
@@ -0,0 +1,108 @@
"""Active-bottle enumeration for the smolmachines backend (PRD
0023 chunk 4 follow-up + issue #77).
Returns a list of `ActiveBottle` records — same shape the docker
backend produces — so CLI `list active` and the dashboard agents
pane render both backends through one code path.
A smolmachines bottle is "active" when its smolvm guest is
running. We cross-reference against the per-bottle sidecar
bundle container to populate the `services` field (which daemons
are up in the bundle); without a bundle we still surface the VM
so the operator can see + clean it up."""
from __future__ import annotations
import json
import shutil
import subprocess
from .. import ActiveBottle
from ..docker.bottle_state import read_metadata
from . import sidecar_bundle as _bundle
from . import smolvm as _smolvm
# Smolvm VM names produced by prepare are `claude-bottle-<slug>`,
# matching the bundle container name pattern. We use the prefix
# both as a filter and to strip back to the slug.
_VM_NAME_PREFIX = "claude-bottle-"
def enumerate_active() -> list[ActiveBottle]:
"""All currently-running smolmachines-backed bottles. Empty
list when smolvm isn't on PATH or no matching VMs are
running."""
if not _smolvm.is_available():
return []
result = subprocess.run(
["smolvm", "machine", "ls", "--json"],
capture_output=True, text=True, check=False,
)
if result.returncode != 0:
return []
try:
machines = json.loads(result.stdout or "[]")
except json.JSONDecodeError:
return []
services_by_slug = _query_bundle_services()
out: list[ActiveBottle] = []
for m in machines:
name = m.get("name") or ""
state = m.get("state") or ""
if state != "running" or not name.startswith(_VM_NAME_PREFIX):
continue
slug = name[len(_VM_NAME_PREFIX):]
metadata = read_metadata(slug)
out.append(ActiveBottle(
backend_name="smolmachines",
slug=slug,
agent_name=metadata.agent_name if metadata else "?",
started_at=metadata.started_at if metadata else "",
services=services_by_slug.get(slug, ()),
))
return out
def _query_bundle_services() -> dict[str, tuple[str, ...]]:
"""`{slug: ('egress', 'pipelock', ...)}` from each running
bundle container's `CLAUDE_BOTTLE_SIDECAR_DAEMONS` env var.
Smolmachines bundles all run the PRD-0024 image with the
same daemon set declared via env, so one inspect per bundle
gets us the picture without exec'ing into the container."""
if shutil.which("docker") is None:
return {}
ps = subprocess.run(
["docker", "ps",
"--filter", "name=" + _bundle.bundle_container_name(""),
"--format", "{{.Names}}"],
capture_output=True, text=True, check=False,
)
if ps.returncode != 0:
return {}
out: dict[str, tuple[str, ...]] = {}
for line in (ps.stdout or "").splitlines():
name = line.strip()
if not name:
continue
slug = name.removeprefix(_bundle.bundle_container_name(""))
if not slug:
continue
inspect = subprocess.run(
["docker", "inspect", name, "--format", "{{json .Config.Env}}"],
capture_output=True, text=True, check=False,
)
if inspect.returncode != 0:
continue
try:
env_list = json.loads(inspect.stdout or "[]")
except json.JSONDecodeError:
continue
for entry in env_list:
key, _, value = entry.partition("=")
if key == "CLAUDE_BOTTLE_SIDECAR_DAEMONS":
out[slug] = tuple(sorted(
d for d in value.split(",") if d
))
break
return out
+104 -83
View File
@@ -26,16 +26,18 @@ from datetime import datetime, timezone
from pathlib import Path
from .. import supervise as _supervise
from ..backend import BottleSpec, get_bottle_backend
from ..backend import (
ActiveBottle,
BottleSpec,
enumerate_active_bottles,
get_bottle_backend,
known_backend_names,
)
from ..backend.docker.capability_apply import (
CapabilityApplyError,
apply_capability_change,
)
from ..backend.docker.bottle_state import bottle_state_dir, read_metadata
from ..backend.docker.compose import (
compose_project_name,
list_active_slugs,
)
from ..backend.docker.egress_apply import (
EgressApplyError,
add_route,
@@ -95,77 +97,22 @@ class QueuedProposal:
queue_dir: Path
@dataclass(frozen=True)
class ActiveAgent:
"""One running bottle, as the agents pane displays it (PRD
0019). `services` is the set of sidecar service names
currently up for this bottle, used to gate which edit verbs
apply (no `egress` → `routes edit` is meaningless)."""
slug: str
agent_name: str # from metadata.json; "?" if missing
started_at: str # ISO 8601 from metadata.json; "" if missing
services: tuple[str, ...] # alphabetical, e.g. ("egress", "pipelock", "supervise")
# `ActiveAgent` was PRD-0019's docker-specific row type. It now
# aliases the shared `ActiveBottle` dataclass so the dashboard
# and the CLI `list active` both render the same source of truth.
# Field surface stays compatible (slug / agent_name / started_at
# / services) plus a new `backend_name` so dashboard rows can
# show which backend a bottle came from.
ActiveAgent = ActiveBottle
def _parse_services_by_project(stdout: str) -> dict[str, set[str]]:
"""Parse `docker ps` output formatted as
`<project-label>\\t<service-label>` (one line per container)
into a `{project: {service, ...}}` mapping. Pure function for
testing — the docker invocation is in the caller."""
out: dict[str, set[str]] = {}
for line in stdout.splitlines():
project, _, service = line.partition("\t")
if not project or not service:
continue
out.setdefault(project, set()).add(service)
return out
def _query_services_by_project() -> dict[str, set[str]]:
"""One `docker ps` call → `{project: {service, ...}}`. PRD
0019 open question #1 picked this shape over per-bottle
`compose ps` calls — for hosts with N bottles, this is one
subprocess instead of N per refresh tick."""
try:
r = subprocess.run(
[
"docker", "ps",
"--filter", "label=com.docker.compose.project",
"--format",
'{{.Label "com.docker.compose.project"}}'
"\t"
'{{.Label "com.docker.compose.service"}}',
],
capture_output=True, text=True, check=False,
)
except FileNotFoundError:
return {}
if r.returncode != 0:
return {}
return _parse_services_by_project(r.stdout or "")
def discover_active_agents() -> list[ActiveAgent]:
"""All currently-running claude-bottle compose projects with
their metadata + service set. Returns [] when docker isn't
reachable. PRD 0019."""
slugs = list_active_slugs()
if not slugs:
return []
services_by_project = _query_services_by_project()
out: list[ActiveAgent] = []
for slug in slugs:
project = compose_project_name(slug)
services = services_by_project.get(project, set())
metadata = read_metadata(slug)
out.append(ActiveAgent(
slug=slug,
agent_name=metadata.agent_name if metadata else "?",
started_at=metadata.started_at if metadata else "",
services=tuple(sorted(services)),
))
return out
def discover_active_agents() -> list[ActiveBottle]:
"""All currently-running bottles across every backend with
their metadata + service set. Returns [] when neither
backend is reachable. Backed by the shared
`enumerate_active_bottles` helper so the CLI's
`./cli.py list active` and this dashboard show the same data."""
return enumerate_active_bottles()
@@ -592,6 +539,68 @@ def _preflight_modal(
return False
def _backend_picker_modal(
stdscr: "curses._CursesWindow",
agent_name: str,
) -> str | None:
"""Modal "which backend to launch this agent on?" picker. Up/
Down + Enter to confirm, Esc / N to abort. Returns the chosen
backend name or None on abort.
Defaults to the first known backend (`docker` lexicographically),
which keeps existing-muscle-memory flows quiet — the modal only
surfaces a choice; it doesn't surprise the operator by jumping
to smolmachines. The picker exists so operators can opt in to
smolmachines without setting CLAUDE_BOTTLE_BACKEND beforehand
(issue #77)."""
names = list(known_backend_names())
if len(names) <= 1:
return names[0] if names else None
selected = 0
h, w = stdscr.getmaxyx()
box_w = min(60, max(20, w - 4))
box_h = min(len(names) + 6, max(8, h - 4))
top = max(0, (h - box_h) // 2)
left = max(0, (w - box_w) // 2)
while True:
win = curses.newwin(box_h, box_w, top, left)
win.erase()
win.box()
win.addnstr(0, 2, " choose backend ", box_w - 4, curses.A_BOLD)
win.addnstr(
1, 2,
f"launching {agent_name!r}; pick a backend:",
box_w - 4,
)
for i, name in enumerate(names):
marker = "" if i == selected else " "
attr = curses.A_REVERSE if i == selected else 0
win.addnstr(3 + i, 2, f"{marker} {name}", box_w - 4, attr)
win.addnstr(
box_h - 2, 2,
" Enter: confirm Esc / N: abort ↑/↓: move ",
box_w - 4, curses.A_DIM,
)
win.refresh()
try:
key = stdscr.getch()
except KeyboardInterrupt:
_erase_modal(stdscr)
return None
if key in (curses.KEY_UP,):
selected = (selected - 1) % len(names)
elif key in (curses.KEY_DOWN,):
selected = (selected + 1) % len(names)
elif key in (curses.KEY_ENTER, 10, 13):
_erase_modal(stdscr)
return names[selected]
elif key in (ord("n"), ord("N"), 27):
_erase_modal(stdscr)
return None
def _erase_modal(stdscr: "curses._CursesWindow") -> None:
"""Force-redraw the dashboard's pre-modal frame so a modal
sub-window's content stops showing. Curses tracks the modal
@@ -1119,6 +1128,13 @@ def _new_agent_flow(
if picked is None:
return "agent start aborted"
# Backend picker (issue #77): operator chooses docker /
# smolmachines per launch. With only one backend installed
# the modal short-circuits (no need to ask).
backend_name = _backend_picker_modal(stdscr, picked)
if backend_name is None:
return f"start of {picked!r} aborted at backend select"
spec = BottleSpec(
manifest=manifest,
agent_name=picked,
@@ -1144,12 +1160,13 @@ def _new_agent_flow(
stage_dir=stage_dir,
render_preflight=_render,
prompt_yes=_prompt,
backend_name=backend_name,
)
if plan is None:
settle_state(identity)
return f"start of {picked!r} aborted at preflight"
backend = get_bottle_backend()
backend = get_bottle_backend(backend_name)
# PRD 0021 follow-up: in tmux, route the launch step's
# stderr (Python info() + subprocess inheritors) into
@@ -1714,21 +1731,25 @@ def _selected_agent(
def _format_agent_row(a: ActiveAgent, maxw: int) -> str:
"""One-line agent row: ` <slug> <agent_name> started <HH:MM:SS>
[<sidecars>]`. The `agent` service is filtered out of the
displayed list — it's always present for an active bottle, so
listing it carries no information; the sidecars are the
differentiator. Truncated to `maxw` because the renderer's
addnstr only enforces width if we hand it a properly-sized
string."""
"""One-line agent row: ` [<backend>] <slug> <agent_name> started
<HH:MM:SS> [<sidecars>]`. The `agent` service is filtered out of
the displayed list — it's always present for an active bottle,
so listing it carries no information; the sidecars are the
differentiator.
The `[docker]` / `[smolmachines]` prefix lets the operator tell
which backend a bottle came from (issue #77). Truncated to
`maxw` because the renderer's addnstr only enforces width if
we hand it a properly-sized string."""
started = (
a.started_at.split("T", 1)[1][:8]
if "T" in a.started_at else (a.started_at or "?")
)
sidecars = tuple(s for s in a.services if s != "agent")
services = ",".join(sidecars) if sidecars else "(starting)"
backend_tag = f"[{a.backend_name}]" if a.backend_name else ""
line = (
f" {a.slug} {a.agent_name} "
f" {backend_tag} {a.slug} {a.agent_name} "
f"started {started} [{services}]"
)
if len(line) > maxw:
+15 -2
View File
@@ -3,8 +3,9 @@
from __future__ import annotations
import argparse
import sys
from ..backend import get_bottle_backend
from ..backend import enumerate_active_bottles
from ..manifest import Manifest
from ._common import PROG, USER_CWD
@@ -20,5 +21,17 @@ def cmd_list(argv: list[str]) -> int:
print(name)
return 0
get_bottle_backend().list_active()
# `active` enumerates every backend (docker + smolmachines)
# so smolmachines bottles aren't hidden behind the env var.
active = enumerate_active_bottles()
if not active:
print("no active claude-bottle bottles", file=sys.stderr)
return 0
# One line per bottle: `<backend>\t<slug>\t<agent>\t<status>`.
# Tab-separated keeps the format stable for shell pipelines;
# the dashboard renders the same data through its own
# formatter.
for b in active:
services = ",".join(b.services) if b.services else "-"
print(f"{b.backend_name}\t{b.slug}\t{b.agent_name}\t{services}")
return 0
+26 -3
View File
@@ -18,7 +18,12 @@ import tempfile
from pathlib import Path
from typing import Callable
from ..backend import Bottle, BottleSpec, get_bottle_backend
from ..backend import (
Bottle,
BottleSpec,
get_bottle_backend,
known_backend_names,
)
from ..backend.docker.bottle_plan import DockerBottlePlan
from ..backend.docker.bottle_state import (
cleanup_state,
@@ -36,6 +41,15 @@ def cmd_start(argv: list[str]) -> int:
parser.add_argument("--dry-run", action="store_true")
parser.add_argument("--cwd", action="store_true", help="copy host cwd into a derived image")
parser.add_argument("--remote-control", action="store_true")
parser.add_argument(
"--backend",
choices=known_backend_names(),
default=None,
help=(
"backend to launch the bottle on (default: $CLAUDE_BOTTLE_BACKEND "
"or 'docker'). Overrides the env var when set."
),
)
parser.add_argument("name", help="agent name defined in claude-bottle.json")
args = parser.parse_args(argv)
@@ -52,6 +66,7 @@ def cmd_start(argv: list[str]) -> int:
spec,
dry_run=dry_run,
remote_control=args.remote_control,
backend_name=args.backend,
)
@@ -65,18 +80,24 @@ def prepare_with_preflight(
render_preflight: Callable[[DockerBottlePlan], None],
prompt_yes: Callable[[], bool],
dry_run: bool = False,
backend_name: str | None = None,
) -> tuple[DockerBottlePlan | None, str]:
"""Run `backend.prepare`, render the preflight summary via the
injected callable, prompt y/N via the injected callable. The CLI
binds these to stderr/stdin; the dashboard binds them to a
curses modal.
`backend_name` selects which backend prepares the plan
(`None` `$CLAUDE_BOTTLE_BACKEND` `docker`). Dashboard
passes the value from its new-agent backend-picker modal; the
CLI passes whatever `--backend` resolved to.
Returns `(plan, identity)`. `plan` is None on dry-run or
operator-N, but `identity` is set as soon as `backend.prepare`
returns so callers can reap the prepare-time state dir via
`settle_state(identity)` in their finally exactly the existing
semantics."""
backend = get_bottle_backend()
backend = get_bottle_backend(backend_name)
plan = backend.prepare(spec, stage_dir=stage_dir)
identity = _identity_from_plan(plan)
@@ -175,6 +196,7 @@ def _launch_bottle(
*,
dry_run: bool,
remote_control: bool,
backend_name: str | None = None,
) -> int:
"""Shared launch core for `start` and `resume`. Builds the plan,
prints / dry-runs / prompts as appropriate, brings the bottle up,
@@ -188,11 +210,12 @@ def _launch_bottle(
render_preflight=_text_render_preflight(remote_control=remote_control),
prompt_yes=_text_prompt_yes,
dry_run=dry_run,
backend_name=backend_name,
)
if plan is None:
return 0
backend = get_bottle_backend()
backend = get_bottle_backend(backend_name)
with backend.launch(plan) as bottle:
exit_code = attach_claude(bottle, remote_control=remote_control)
info(
+93
View File
@@ -0,0 +1,93 @@
"""Unit: backend selection + cross-backend enumeration (issue #77).
`get_bottle_backend(name)` resolves a backend by explicit name,
env var, or default. `enumerate_active_bottles()` walks every
registered backend and concatenates their `ActiveBottle`
listings the CLI and dashboard both go through this so adding
a backend lights it up in both places."""
from __future__ import annotations
import os
import unittest
from unittest.mock import patch
from claude_bottle import backend as backend_mod
from claude_bottle.backend import (
ActiveBottle,
enumerate_active_bottles,
get_bottle_backend,
known_backend_names,
)
class TestGetBottleBackend(unittest.TestCase):
def test_explicit_name_wins_over_env(self):
with patch.dict(os.environ, {"CLAUDE_BOTTLE_BACKEND": "smolmachines"}):
b = get_bottle_backend("docker")
self.assertEqual("docker", b.name)
def test_env_var_fallback(self):
with patch.dict(os.environ, {"CLAUDE_BOTTLE_BACKEND": "smolmachines"}):
b = get_bottle_backend()
self.assertEqual("smolmachines", b.name)
def test_default_docker(self):
with patch.dict(os.environ, {}, clear=True):
b = get_bottle_backend()
self.assertEqual("docker", b.name)
def test_unknown_dies(self):
with patch.object(backend_mod, "die", side_effect=SystemExit("die")):
with self.assertRaises(SystemExit):
get_bottle_backend("nonexistent")
class TestKnownBackendNames(unittest.TestCase):
def test_returns_both_backends_sorted(self):
self.assertEqual(("docker", "smolmachines"), known_backend_names())
class TestEnumerateActiveBottles(unittest.TestCase):
"""Combines each backend's `enumerate_active`. Each backend's
implementation has its own tests (`test_docker_enumerate_active`,
`test_smolmachines_*`); this just asserts the aggregator stitches
them together."""
def test_concatenates_per_backend(self):
a = ActiveBottle(
backend_name="docker", slug="a-1", agent_name="impl",
started_at="", services=("pipelock",),
)
b = ActiveBottle(
backend_name="smolmachines", slug="b-2", agent_name="research",
started_at="", services=(),
)
class _FakeBackend:
def __init__(self, items):
self._items = items
def enumerate_active(self):
return self._items
with patch.object(
backend_mod, "_BACKENDS",
{"docker": _FakeBackend([a]), "smolmachines": _FakeBackend([b])},
):
self.assertEqual([a, b], enumerate_active_bottles())
def test_empty_when_no_backends_have_active(self):
class _FakeBackend:
def enumerate_active(self):
return []
with patch.object(
backend_mod, "_BACKENDS",
{"docker": _FakeBackend(), "smolmachines": _FakeBackend()},
):
self.assertEqual([], enumerate_active_bottles())
if __name__ == "__main__":
unittest.main()
+61
View File
@@ -0,0 +1,61 @@
"""Unit: `cli.py start --backend=<name>` flag (issue #77).
Asserts that the flag wins over the env var, that the env var is
the fallback, and that the choices are pulled from the backend
registry (so adding a backend lights up in argparse without code
edits)."""
from __future__ import annotations
import argparse
import os
import unittest
from unittest.mock import patch
from claude_bottle.backend import known_backend_names
class TestStartBackendFlag(unittest.TestCase):
"""The flag is wired by `cmd_start`'s argparse and threaded
through `prepare_with_preflight(backend_name=...)`. Rather than
drive the whole start flow (which builds containers), we test
the argparse shape and the resolution function separately."""
def _build_parser(self):
# Mirror the parser definition from `cmd_start` so this
# test doesn't have to invoke the full command.
parser = argparse.ArgumentParser(prog="cb start")
parser.add_argument(
"--backend",
choices=known_backend_names(),
default=None,
)
parser.add_argument("name")
return parser
def test_flag_recognized(self):
args = self._build_parser().parse_args(["--backend=smolmachines", "researcher"])
self.assertEqual("smolmachines", args.backend)
self.assertEqual("researcher", args.name)
def test_flag_default_none_means_env_or_docker(self):
args = self._build_parser().parse_args(["researcher"])
self.assertIsNone(args.backend)
def test_invalid_backend_rejected_by_argparse(self):
parser = self._build_parser()
with self.assertRaises(SystemExit):
parser.parse_args(["--backend=garbage", "researcher"])
def test_resolution_priority_explicit_over_env(self):
# Independent assertion that get_bottle_backend (where
# `--backend` ultimately threads to) prefers the explicit
# name over CLAUDE_BOTTLE_BACKEND.
from claude_bottle.backend import get_bottle_backend
with patch.dict(os.environ, {"CLAUDE_BOTTLE_BACKEND": "smolmachines"}):
self.assertEqual("docker", get_bottle_backend("docker").name)
if __name__ == "__main__":
unittest.main()
+12 -145
View File
@@ -1,18 +1,10 @@
"""Unit: dashboard.discover_active_agents (PRD 0019 chunk 1).
"""Unit: dashboard's row-formatting + selection helpers (PRD 0019).
The full discover function fans out to `docker compose ls`, `docker
ps`, and per-bottle metadata.json reads too much for a unit test.
Tests split into:
- Parser tests for `_parse_services_by_project`: pure function, no
I/O, deterministic on its input string.
- Integration-shaped tests that monkeypatch the slug list +
services map and read metadata from a fake home, then assert
the assembled `ActiveAgent` shape.
The actual `docker ps` invocation is exercised by manual probing
during development and the (real-docker) integration tests; here
we lock down the shape contract so a regression surfaces in unit CI.
The active-bottle enumeration tests moved to
`test_docker_enumerate_active.py` (issue #77) — dashboard now
delegates listing to `enumerate_active_bottles` so the parser
and assembly tests live next to the docker backend's
implementation.
"""
from __future__ import annotations
@@ -22,54 +14,9 @@ import unittest
from pathlib import Path
from claude_bottle import supervise
from claude_bottle.backend.docker import bottle_state
from claude_bottle.cli import dashboard
class TestParseServicesByProject(unittest.TestCase):
def test_empty_input(self):
self.assertEqual({}, dashboard._parse_services_by_project(""))
def test_one_container(self):
out = dashboard._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
)
self.assertEqual({"claude-bottle-dev-abc": {"egress"}}, out)
def test_multiple_services_per_project(self):
out = dashboard._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
"claude-bottle-dev-abc\tpipelock\n"
"claude-bottle-dev-abc\tsupervise\n"
)
self.assertEqual(
{"claude-bottle-dev-abc": {"egress", "pipelock", "supervise"}},
out,
)
def test_multiple_projects(self):
out = dashboard._parse_services_by_project(
"proj-a\tegress\n"
"proj-b\tpipelock\n"
"proj-a\tsupervise\n"
)
self.assertEqual(
{"proj-a": {"egress", "supervise"}, "proj-b": {"pipelock"}},
out,
)
def test_skips_lines_missing_either_field(self):
# Defends against unlabeled containers slipping into the
# output (the filter should prevent it, but be robust).
out = dashboard._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
"no-tab-here\n"
"\tmissing-project\n"
"missing-service\t\n"
)
self.assertEqual({"claude-bottle-dev-abc": {"egress"}}, out)
class _FakeHomeMixin:
def _setup_fake_home(self) -> None:
self._tmp = tempfile.TemporaryDirectory(prefix="dashboard-aa-test.")
@@ -86,97 +33,12 @@ class _FakeHomeMixin:
self._tmp.cleanup()
class TestDiscoverActiveAgents(_FakeHomeMixin, unittest.TestCase):
def setUp(self) -> None:
self._setup_fake_home()
self._orig_slugs = dashboard.list_active_slugs
self._orig_services = dashboard._query_services_by_project
def tearDown(self) -> None:
dashboard.list_active_slugs = self._orig_slugs
dashboard._query_services_by_project = self._orig_services
self._teardown_fake_home()
def _stub(self, slugs: list[str], services_by_project: dict[str, set[str]]) -> None:
dashboard.list_active_slugs = lambda: slugs
dashboard._query_services_by_project = lambda: services_by_project
def test_no_active_slugs_returns_empty(self):
self._stub([], {})
self.assertEqual([], dashboard.discover_active_agents())
def test_assembles_from_metadata_and_services(self):
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity="dev-abc",
agent_name="implementer",
cwd="",
copy_cwd=False,
started_at="2026-05-26T03:00:00+00:00",
compose_project="claude-bottle-dev-abc",
))
self._stub(
["dev-abc"],
{"claude-bottle-dev-abc": {"pipelock", "egress", "supervise"}},
)
agents = dashboard.discover_active_agents()
self.assertEqual(1, len(agents))
a = agents[0]
self.assertEqual("dev-abc", a.slug)
self.assertEqual("implementer", a.agent_name)
self.assertEqual("2026-05-26T03:00:00+00:00", a.started_at)
self.assertEqual(("egress", "pipelock", "supervise"), a.services)
def test_missing_metadata_renders_question_mark(self):
# State dir doesn't exist for this slug — agent_name falls
# back to "?" rather than dropping the row.
self._stub(["mystery-zzz"], {"claude-bottle-mystery-zzz": {"pipelock"}})
agents = dashboard.discover_active_agents()
self.assertEqual(1, len(agents))
self.assertEqual("?", agents[0].agent_name)
self.assertEqual("", agents[0].started_at)
self.assertEqual(("pipelock",), agents[0].services)
def test_no_services_for_project_yields_empty_tuple(self):
# Race window between `compose up` returning and the actual
# containers being listed in `docker ps` — render the row
# but with no services.
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity="warming-up",
agent_name="researcher",
cwd="",
copy_cwd=False,
started_at="2026-05-26T03:05:00+00:00",
compose_project="claude-bottle-warming-up",
))
self._stub(["warming-up"], {})
agents = dashboard.discover_active_agents()
self.assertEqual((), agents[0].services)
def test_preserves_slug_order(self):
for slug in ("z-1", "a-1", "m-1"):
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity=slug,
agent_name=slug.split("-")[0],
cwd="",
copy_cwd=False,
started_at="t",
compose_project=f"claude-bottle-{slug}",
))
# list_active_slugs returns sorted; preserve that order in
# the output.
self._stub(["a-1", "m-1", "z-1"], {})
agents = dashboard.discover_active_agents()
self.assertEqual(
["a-1", "m-1", "z-1"],
[a.slug for a in agents],
)
class TestFormatAgentRow(unittest.TestCase):
"""One-line row formatting for the agents pane (PRD 0019 chunk 2)."""
def _agent(self, **overrides) -> dashboard.ActiveAgent:
defaults = dict(
backend_name="docker",
slug="dev-abc12",
agent_name="implementer",
started_at="2026-05-26T02:55:01+00:00",
@@ -232,6 +94,7 @@ class TestSelectionStatus(unittest.TestCase):
def _agent(self, slug: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name="x", started_at="", services=(),
)
@@ -295,6 +158,7 @@ class TestRunningCounts(unittest.TestCase):
def _agent(self, agent_name: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=f"{agent_name}-abc",
agent_name=agent_name,
started_at="",
@@ -329,6 +193,7 @@ class TestSelectedAgent(unittest.TestCase):
def _agent(self, slug: str, services: tuple[str, ...] = ()) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name="x", started_at="", services=services,
)
@@ -387,6 +252,7 @@ class TestPickNextAfterStop(unittest.TestCase):
def _agent(self, slug: str) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug=slug, agent_name=slug, started_at="", services=(),
)
@@ -579,6 +445,7 @@ class TestOperatorEditFlowGuards(_FakeHomeMixin, unittest.TestCase):
def _agent(self, services: tuple[str, ...]) -> dashboard.ActiveAgent:
return dashboard.ActiveAgent(
backend_name="docker",
slug="dev-abc12",
agent_name="impl",
started_at="",
+194
View File
@@ -0,0 +1,194 @@
"""Unit: docker backend's `enumerate_active` (issue #77).
The full enumerate function fans out to `docker compose ls`,
`docker ps`, and per-bottle metadata.json reads too much for a
unit test. Tests split into:
- Parser tests for `_parse_services_by_project`: pure function,
no I/O, deterministic on its input string.
- Integration-shaped tests that monkeypatch the slug list +
services map and read metadata from a fake home, then assert
the assembled `ActiveBottle` shape.
The actual `docker ps` invocation is exercised by manual probing
during development and the (real-docker) integration tests; here
we lock down the shape contract so a regression surfaces in unit
CI. Tests moved out of `test_dashboard_active_agents.py` as part
of issue #77 — the dashboard now delegates to this layer.
"""
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, cleanup
class TestParseServicesByProject(unittest.TestCase):
def test_empty_input(self):
self.assertEqual({}, cleanup._parse_services_by_project(""))
def test_one_container(self):
out = cleanup._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
)
self.assertEqual({"claude-bottle-dev-abc": {"egress"}}, out)
def test_multiple_services_per_project(self):
out = cleanup._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
"claude-bottle-dev-abc\tpipelock\n"
"claude-bottle-dev-abc\tsupervise\n"
)
self.assertEqual(
{"claude-bottle-dev-abc": {"egress", "pipelock", "supervise"}},
out,
)
def test_multiple_projects(self):
out = cleanup._parse_services_by_project(
"proj-a\tegress\n"
"proj-b\tpipelock\n"
"proj-a\tsupervise\n"
)
self.assertEqual(
{"proj-a": {"egress", "supervise"}, "proj-b": {"pipelock"}},
out,
)
def test_skips_lines_missing_either_field(self):
# Defends against unlabeled containers slipping into the
# output (the filter should prevent it, but be robust).
out = cleanup._parse_services_by_project(
"claude-bottle-dev-abc\tegress\n"
"no-tab-here\n"
"\tmissing-project\n"
"missing-service\t\n"
)
self.assertEqual({"claude-bottle-dev-abc": {"egress"}}, out)
class _FakeHomeMixin:
def _setup_fake_home(self) -> None:
self._tmp = tempfile.TemporaryDirectory(prefix="enum-active.")
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_home = lambda: setattr(supervise, "claude_bottle_root", original)
def _teardown_fake_home(self) -> None:
self._restore_home()
self._tmp.cleanup()
class TestEnumerateActive(_FakeHomeMixin, unittest.TestCase):
def setUp(self) -> None:
self._setup_fake_home()
self._orig_slugs = cleanup.list_active_slugs
self._orig_services = cleanup._query_services_by_project
# Skip the docker-availability gate so tests don't need a
# real docker on PATH.
import shutil
self._orig_which = shutil.which
shutil.which = lambda name: "/usr/bin/" + name if name == "docker" else None
def tearDown(self) -> None:
cleanup.list_active_slugs = self._orig_slugs
cleanup._query_services_by_project = self._orig_services
import shutil
shutil.which = self._orig_which
self._teardown_fake_home()
def _stub(self, slugs: list[str], services_by_project: dict[str, set[str]]) -> None:
cleanup.list_active_slugs = lambda **_: slugs
cleanup._query_services_by_project = lambda: services_by_project
def test_no_active_slugs_returns_empty(self):
self._stub([], {})
self.assertEqual([], cleanup.enumerate_active())
def test_assembles_from_metadata_and_services(self):
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity="dev-abc",
agent_name="implementer",
cwd="",
copy_cwd=False,
started_at="2026-05-26T03:00:00+00:00",
compose_project="claude-bottle-dev-abc",
))
self._stub(
["dev-abc"],
{"claude-bottle-dev-abc": {"pipelock", "egress", "supervise"}},
)
active = cleanup.enumerate_active()
self.assertEqual(1, len(active))
a = active[0]
self.assertEqual("docker", a.backend_name)
self.assertEqual("dev-abc", a.slug)
self.assertEqual("implementer", a.agent_name)
self.assertEqual("2026-05-26T03:00:00+00:00", a.started_at)
self.assertEqual(("egress", "pipelock", "supervise"), a.services)
def test_missing_metadata_renders_question_mark(self):
# State dir doesn't exist for this slug — agent_name falls
# back to "?" rather than dropping the row.
self._stub(["mystery-zzz"], {"claude-bottle-mystery-zzz": {"pipelock"}})
active = cleanup.enumerate_active()
self.assertEqual(1, len(active))
self.assertEqual("?", active[0].agent_name)
self.assertEqual("", active[0].started_at)
self.assertEqual(("pipelock",), active[0].services)
def test_no_services_for_project_yields_empty_tuple(self):
# Race window between `compose up` returning and the actual
# containers being listed in `docker ps` — render the row
# but with no services.
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity="warming-up",
agent_name="researcher",
cwd="",
copy_cwd=False,
started_at="2026-05-26T03:05:00+00:00",
compose_project="claude-bottle-warming-up",
))
self._stub(["warming-up"], {})
active = cleanup.enumerate_active()
self.assertEqual((), active[0].services)
def test_preserves_slug_order(self):
for slug in ("z-1", "a-1", "m-1"):
bottle_state.write_metadata(bottle_state.BottleMetadata(
identity=slug,
agent_name=slug.split("-")[0],
cwd="",
copy_cwd=False,
started_at="t",
compose_project=f"claude-bottle-{slug}",
))
# list_active_slugs returns sorted; preserve that order in
# the output.
self._stub(["a-1", "m-1", "z-1"], {})
active = cleanup.enumerate_active()
self.assertEqual(
["a-1", "m-1", "z-1"],
[a.slug for a in active],
)
def test_noop_when_docker_missing(self):
# Defensive: list active shouldn't die just because docker
# isn't on PATH (smolmachines bottles are still discoverable
# via the other backend's enumerate).
import shutil
shutil.which = lambda _name: None
self._stub(["some-slug"], {})
self.assertEqual([], cleanup.enumerate_active())
if __name__ == "__main__":
unittest.main()