refactor(cli): group subcommand handlers under cli/commands/

Move the eleven per-command modules (backend, cleanup, commit, edit,
info, init, list, login, resume, start, supervise) into a new
bot_bottle/cli/commands/ package, leaving the dispatcher (__init__),
entrypoint (__main__), and shared helpers (_common, tui) at the cli
root. Makes the command surface obvious at a glance and separates
handlers from the plumbing that registers them.

Updated the dispatcher's COMMANDS imports to .commands.*, bumped the
moved files' relative-import depths (.._common, .. import tui, sibling
.start unchanged), and repointed test references to
bot_bottle.cli.commands.*. Full unit suite green (2243); CLI dispatch
verified via `python -m bot_bottle.cli --help`.

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

bot_bottle/cli/commands/__init__.py

@@ -0,0 +1,6 @@
+"""CLI subcommand handlers — one module per `bot-bottle <command>`.
+
+Each module exposes a `cmd_<name>(argv)` handler that the dispatcher
+(`bot_bottle.cli`) registers in its COMMANDS table. Shared CLI helpers
+(`_common`, `tui`) stay one level up in the `cli` package.
+"""

bot_bottle/cli/commands/backend.py

@@ -0,0 +1,46 @@
+"""`backend` CLI command — generic host setup/status across backends.
+
+`./cli.py backend setup    [--backend=NAME]` provisions (or points at how
+to provision) the chosen backend's one-time host prerequisites.
+`./cli.py backend status   [--backend=NAME]` reports readiness.
+`./cli.py backend teardown [--backend=NAME]` undoes setup (uninstall).
+
+All dispatch to the backend's `setup()` / `status()` / `teardown()`
+classmethods, so there are no backend-specific commands — swapping
+backends is just a different `--backend` (or `$BOT_BOTTLE_BACKEND`, or
+the host default).
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from ...backend import get_bottle_backend, known_backend_names
+from .._common import PROG
+
+
+def cmd_backend(args: list[str]) -> int:
+    parser = argparse.ArgumentParser(
+        prog=f"{PROG} backend",
+        description="Set up or check a backend's host prerequisites.",
+    )
+    parser.add_argument(
+        "action",
+        choices=("setup", "status", "teardown"),
+        help="setup: provision/print host prerequisites; status: report "
+             "readiness; teardown: undo setup (uninstall)",
+    )
+    parser.add_argument(
+        "--backend",
+        choices=known_backend_names(),
+        default=None,
+        help="backend to target (default: $BOT_BOTTLE_BACKEND or the host default)",
+    )
+    ns = parser.parse_args(args)
+
+    backend = get_bottle_backend(ns.backend)
+    if ns.action == "setup":
+        return backend.setup()
+    if ns.action == "teardown":
+        return backend.teardown()
+    return backend.status()

bot_bottle/cli/commands/cleanup.py

@@ -0,0 +1,67 @@
+"""cleanup: stop and remove all orphaned bot-bottle resources.
+
+Walks every registered backend (docker, firecracker, macos-container)
+so a single `./cli.py cleanup` reaps every backend's leftovers — a
+firecracker bottle's VM processes and run dirs 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 non-docker 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 preserved sessions the operator may want to `resume`. Manual
+`rm -rf ~/.bot-bottle/state/<identity>` is the path for those.
+"""
+
+from __future__ import annotations
+
+import sys
+
+from ...backend import get_bottle_backend, has_backend, known_backend_names
+from ...log import info
+from .._common import read_tty_line
+
+
+def cmd_cleanup(_argv: list[str]) -> int:
+    # Order: stable backend iteration so the y/N output is
+    # deterministic across runs. Skip backends whose runtime
+    # isn't available on this host so e.g. macos-container
+    # doesn't error on Linux.
+    plans = [
+        (name, get_bottle_backend(name))
+        for name in known_backend_names()
+        if has_backend(name)
+    ]
+    prepared = [(name, b, b.prepare_cleanup()) for name, b in plans]
+
+    if all(p.empty for _, _, p in prepared):
+        info("no bot-bottle resources to clean up")
+        return 0
+
+    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
+
+    for name, backend, plan in prepared:
+        if plan.empty:
+            continue
+        backend.cleanup(plan)
+    info("cleanup: done")
+    return 0
+
+
+def _prompt_yes(message: str) -> bool:
+    sys.stderr.write(f"bot-bottle: {message} [y/N] ")
+    sys.stderr.flush()
+    reply = read_tty_line()
+    return reply in ("y", "Y", "yes", "YES")

bot_bottle/cli/commands/commit.py

@@ -0,0 +1,53 @@
+"""commit: freeze a running bottle's state to a resumable artifact.
+
+Docker bottles are committed to a local Docker image. Macos-container
+bottles are exported and rebuilt as a local Apple Container image.
+Firecracker bottles stream the guest rootfs out over SSH and rebuild a
+local Docker image. The resulting reference is stored in per-bottle
+state so the next `./cli.py resume <slug>` boots from the snapshot
+instead of rebuilding from the Dockerfile.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from ...backend import enumerate_active_agents
+from ...backend.freeze import CommitCancelled, get_freezer
+from ...bottle_state import read_metadata
+from ...log import die
+from .._common import PROG
+from .. import tui
+
+
+def cmd_commit(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} commit", add_help=True)
+    parser.add_argument(
+        "slug",
+        nargs="?",
+        default=None,
+        help=(
+            "bottle slug from `cli.py list active` "
+            "(omit to pick interactively)"
+        ),
+    )
+    args = parser.parse_args(argv)
+
+    slug = args.slug
+    if slug is None:
+        active = enumerate_active_agents()
+        if not active:
+            die("no active bottles; start one with `./cli.py start`")
+        choices = [a.slug for a in active]
+        slug = tui.filter_select(choices, title="Select bottle to commit")
+        if slug is None:
+            return 0
+
+    metadata = read_metadata(slug)
+    backend = metadata.backend if metadata else ""
+
+    try:
+        get_freezer(backend).commit_slug(slug)
+    except CommitCancelled:
+        return 0
+    return 0

bot_bottle/cli/commands/edit.py

@@ -0,0 +1,43 @@
+"""edit: open an agent in vim for editing."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+from pathlib import Path
+
+from ...log import die
+from .._common import PROG, USER_CWD
+
+
+def cmd_edit(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} edit", add_help=True)
+    parser.add_argument("scope", choices=["user", "project"])
+    parser.add_argument("name")
+    args = parser.parse_args(argv)
+
+    if args.scope == "user":
+        target_file = Path(os.environ["HOME"]) / "bot-bottle.json"
+    else:
+        target_file = Path(USER_CWD) / "bot-bottle.json"
+
+    if not target_file.is_file():
+        die(f"{target_file} does not exist")
+
+    try:
+        doc = json.loads(target_file.read_text())
+    except json.JSONDecodeError:
+        die(f"{target_file} is not valid JSON")
+
+    if args.name not in (doc.get("agents") or {}):
+        die(f"agent '{args.name}' not found in {target_file}")
+
+    line = 1
+    text = target_file.read_text().splitlines()
+    needle = f'"{args.name}"'
+    for idx, l in enumerate(text, start=1):
+        if needle in l:
+            line = idx
+            break
+    os.execvp("vim", ["vim", f"+{line}", str(target_file)])

bot_bottle/cli/commands/info.py

@@ -0,0 +1,49 @@
+"""info: print env, skills, and prompt details for a named agent."""
+
+from __future__ import annotations
+
+import argparse
+
+from ...log import info
+from ...manifest import ManifestIndex
+from .._common import PROG, USER_CWD
+
+
+def cmd_info(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} info", add_help=True)
+    parser.add_argument("name", help="agent name defined in bot-bottle.json")
+    args = parser.parse_args(argv)
+
+    names = ManifestIndex.resolve(USER_CWD)
+    names.require_agent(args.name)
+    manifest = names.load_for_agent(args.name)
+
+    agent = manifest.agent
+    bottle = manifest.bottle
+    env_names = list(bottle.env.keys())
+    prompt_first_line = agent.prompt.splitlines()[0] if agent.prompt else ""
+
+    print()
+    info(f"agent           : {args.name}")
+    info(f"env (names only): {', '.join(env_names) if env_names else '(none)'}")
+    info(f"skills          : {' '.join(agent.skills) if agent.skills else '(none)'}")
+    info(
+        f"prompt          : {len(agent.prompt)} chars; "
+        f"first line: {prompt_first_line or '(empty)'}"
+    )
+    info(f"bottle             : {agent.bottle}")
+    identity = manifest.git_identity_summary()
+    if identity:
+        info(f"  git identity  : {identity}")
+    if bottle.git:
+        for e in bottle.git:
+            info(
+                f"  git remote    : {e.Name} -> {e.Upstream} "
+                f"(IdentityFile={e.IdentityFile})"
+            )
+            if e.KnownHostKey:
+                info(f"    KnownHostKey: {e.KnownHostKey}")
+    else:
+        info("  git remotes   : (none)")
+    print()
+    return 0

bot_bottle/cli/commands/init.py

@@ -0,0 +1,220 @@
+"""init: interactively create a new agent and add it to bot-bottle.json."""
+
+from __future__ import annotations
+
+import argparse
+import json
+import os
+import re
+import sys
+from pathlib import Path
+from typing import Any
+
+from ...log import die, info, warn
+from .._common import PROG, USER_CWD, read_tty_line
+
+
+def cmd_init(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} init", add_help=True)
+    parser.add_argument("scope", choices=["user", "project"])
+    args = parser.parse_args(argv)
+
+    if args.scope == "user":
+        target_file = Path(os.environ["HOME"]) / "bot-bottle.json"
+    else:
+        target_file = Path(USER_CWD) / "bot-bottle.json"
+
+    print(file=sys.stderr)
+    info(f"bot-bottle init — adding a new agent to {target_file}")
+    print(file=sys.stderr)
+
+    # Agent name
+    agent_name = ""
+    while not agent_name:
+        sys.stderr.write("Agent name: ")
+        sys.stderr.flush()
+        agent_name = read_tty_line().replace(" ", "-")
+        if not agent_name:
+            warn("agent name cannot be empty")
+
+    if not re.match(r"^[a-z0-9][a-z0-9-]*$", agent_name):
+        warn(
+            f"agent name '{agent_name}' contains non-slug characters; "
+            f"it will still work but may cause container naming issues"
+        )
+
+    existing: dict[str, Any] = {}
+    if target_file.is_file():
+        try:
+            existing = json.loads(target_file.read_text())
+        except json.JSONDecodeError:
+            die(f"{target_file} exists but is not valid JSON; fix or remove it first")
+        if agent_name in (existing.get("agents") or {}):
+            sys.stderr.write(
+                f'bot-bottle: agent "{agent_name}" already exists in '
+                f'{target_file}. Overwrite? [y/N] '
+            )
+            sys.stderr.flush()
+            ow = read_tty_line()
+            if ow not in ("y", "Y", "yes", "YES"):
+                info("aborted")
+                return 0
+
+    # Skills
+    print(file=sys.stderr)
+    sys.stderr.write("Skills (space or comma separated, or Enter for none): ")
+    sys.stderr.flush()
+    skills_input = read_tty_line()
+    skill_list: list[str] = []
+    if skills_input:
+        cleaned = skills_input.replace(",", " ")
+        skill_list = [s for s in cleaned.split() if s]
+
+    # Prompt
+    print(file=sys.stderr)
+    info(
+        "System prompt — enter text, then a lone '.' on its own line to "
+        "finish (just '.' to leave empty):"
+    )
+    prompt_lines: list[str] = []
+    while True:
+        line = read_tty_line()
+        if line == ".":
+            break
+        prompt_lines.append(line)
+    prompt_content = "\n".join(prompt_lines)
+
+    # Bottle association
+    print(file=sys.stderr)
+    sys.stderr.write("Associate this agent with a bottle? [y/N] ")
+    sys.stderr.flush()
+    bottle_yn = read_tty_line()
+    bottle_name = ""
+    bottle_env: dict[str, str] = {}
+    bottle_ssh: list[dict[str, Any]] = []
+    bottle_exists_already = False
+    if bottle_yn in ("y", "Y", "yes", "YES"):
+        while not bottle_name:
+            sys.stderr.write("  Bottle name: ")
+            sys.stderr.flush()
+            bottle_name = read_tty_line().replace(" ", "-")
+            if not bottle_name:
+                warn("bottle name cannot be empty")
+
+        if bottle_name in (existing.get("bottles") or {}):
+            bottle_exists_already = True
+            info(
+                f"Bottle '{bottle_name}' already exists in {target_file}; "
+                f"agent will reference it."
+            )
+        else:
+            info(f"Creating new bottle '{bottle_name}'.")
+            bottle_env = _prompt_for_env_vars()
+            sys.stderr.write("  Add SSH host entries to this bottle? [y/N] ")
+            sys.stderr.flush()
+            ssh_yn = read_tty_line()
+            if ssh_yn in ("y", "Y", "yes", "YES"):
+                bottle_ssh = _prompt_for_ssh_entries()
+
+    # Build agent JSON
+    agent_def: dict[str, Any] = {"skills": skill_list, "prompt": prompt_content}
+    if bottle_name:
+        agent_def["bottle"] = bottle_name
+
+    merged: dict[str, Any] = {
+        "bottles": dict(existing.get("bottles") or {}),
+        "agents": dict(existing.get("agents") or {}),
+    }
+    if bottle_name and not bottle_exists_already:
+        merged["bottles"][bottle_name] = {"env": bottle_env, "ssh": bottle_ssh}
+    merged["agents"][agent_name] = agent_def
+
+    target_file.write_text(json.dumps(merged, indent=2) + "\n")
+    info(f"Agent '{agent_name}' written to {target_file}.")
+    info(f"Run '{PROG} info {agent_name}' to verify.")
+    print(file=sys.stderr)
+    return 0
+
+
+def _prompt_for_env_vars() -> dict[str, str]:
+    print(file=sys.stderr)
+    info(
+        "Env vars — enter each var name then its mode. Press Enter with "
+        "no name to finish."
+    )
+    info(
+        "  Modes:  secret (prompt at runtime) | interpolated (read from "
+        "host env) | literal (hardcoded value)"
+    )
+    out: dict[str, str] = {}
+    while True:
+        print(file=sys.stderr)
+        sys.stderr.write("  Var name (or Enter to finish): ")
+        sys.stderr.flush()
+        vname = read_tty_line()
+        if not vname:
+            break
+        sys.stderr.write("  Mode [secret/interpolated/literal] (default: secret): ")
+        sys.stderr.flush()
+        vmode = read_tty_line() or "secret"
+        if vmode == "secret":
+            sys.stderr.write(f'  Prompt message shown to user (default: "enter {vname}"): ')
+            sys.stderr.flush()
+            smsg = read_tty_line()
+            value = f"?{smsg}" if smsg else "?"
+        elif vmode == "interpolated":
+            sys.stderr.write(f"  Host env var to read from (default: {vname}): ")
+            sys.stderr.flush()
+            hvar = read_tty_line() or vname
+            value = "${" + hvar + "}"
+        elif vmode == "literal":
+            sys.stderr.write("  Value: ")
+            sys.stderr.flush()
+            value = read_tty_line()
+        else:
+            warn(f"unknown mode '{vmode}'; using secret")
+            value = "?"
+        out[vname] = value
+    return out
+
+
+def _prompt_for_ssh_entries() -> list[dict[str, Any]]:
+    out: list[dict[str, Any]] = []
+    while True:
+        print(file=sys.stderr)
+        sys.stderr.write("    SSH Host alias (or Enter to finish): ")
+        sys.stderr.flush()
+        shost = read_tty_line()
+        if not shost:
+            break
+        sys.stderr.write("    Hostname (actual hostname or IP): ")
+        sys.stderr.flush()
+        shostname = read_tty_line()
+        sys.stderr.write("    User: ")
+        sys.stderr.flush()
+        suser = read_tty_line()
+        sys.stderr.write("    Port (default: 22): ")
+        sys.stderr.flush()
+        sport_raw = read_tty_line() or "22"
+        if not sport_raw.isdigit():
+            warn("port must be a number; defaulting to 22")
+            sport_raw = "22"
+        sport = int(sport_raw)
+        sys.stderr.write("    IdentityFile (path to private key on host): ")
+        sys.stderr.flush()
+        sidentity = read_tty_line()
+        sys.stderr.write("    KnownHostKey (optional, Enter to skip): ")
+        sys.stderr.flush()
+        skhk = read_tty_line()
+
+        entry: dict[str, Any] = {
+            "Host": shost,
+            "Hostname": shostname,
+            "User": suser,
+            "Port": sport,
+            "IdentityFile": sidentity,
+        }
+        if skhk:
+            entry["KnownHostKey"] = skhk
+        out.append(entry)
+    return out

bot_bottle/cli/commands/list.py

@@ -0,0 +1,62 @@
+"""list: list available agents or active bottles."""
+
+from __future__ import annotations
+
+import argparse
+import os
+import sys
+
+from ...backend import enumerate_active_agents
+from ...manifest import ManifestIndex
+from .._common import PROG, USER_CWD
+
+_ANSI_COLOR_CODES: dict[str, str] = {
+    "red": "\033[91m",
+    "green": "\033[92m",
+    "yellow": "\033[93m",
+    "blue": "\033[94m",
+    "magenta": "\033[95m",
+}
+_ANSI_RESET = "\033[0m"
+
+
+def _ansi_label(text: str, color: str) -> str:
+    if not color:
+        return text
+    if not sys.stdout.isatty():
+        return text
+    term = os.environ.get("TERM", "")
+    if term in ("dumb", ""):
+        return text
+    code = _ANSI_COLOR_CODES.get(color)
+    if not code:
+        return text
+    return f"{code}{text}{_ANSI_RESET}"
+
+
+def cmd_list(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} list", add_help=True)
+    parser.add_argument("scope", choices=["available", "active"])
+    args = parser.parse_args(argv)
+
+    if args.scope == "available":
+        manifest = ManifestIndex.resolve(USER_CWD)
+        for name in manifest.all_agent_names:
+            print(name)
+        return 0
+
+    # `active` enumerates every backend (docker, firecracker,
+    # macos-container) so non-docker bottles aren't hidden behind
+    # the env var.
+    active = enumerate_active_agents()
+    if not active:
+        print("no active bot-bottle bottles", file=sys.stderr)
+        return 0
+    # One line per bottle: `<backend>\t<slug>\t<label>\t<services>`.
+    # Tab-separated keeps the format stable for shell pipelines.
+    for b in active:
+        services = ",".join(b.services) if b.services else "-"
+        display_name = f"{b.label} ({b.agent_name})" if b.label else b.agent_name
+        colored_name = _ansi_label(display_name, b.color)
+        print(f"{b.backend_name}\t{b.slug}\t{colored_name}\t{services}")
+    return 0

bot_bottle/cli/commands/login.py

@@ -0,0 +1,168 @@
+"""bb login — register this host with a bot-bottle console.
+
+Opens a device-authorization flow against the target console, waits for the
+operator to approve, then writes access and refresh tokens to
+~/.bot-bottle/console.json (or $BOT_BOTTLE_ROOT/console.json).
+
+Usage:
+  bb login [--console-url URL] [--label LABEL]
+
+Flags:
+  --console-url URL   Target console URL (overrides BB_CONSOLE_URL env var)
+  --label LABEL       Host label shown in the console (default: hostname)
+"""
+
+from __future__ import annotations
+
+import json
+import os
+import socket
+import sys
+import tempfile
+import time
+import urllib.error
+import urllib.request
+from pathlib import Path
+from typing import Any
+
+from ...paths import bot_bottle_root
+
+_CONSOLE_URL_ENV = "BB_CONSOLE_URL"
+_POLL_SLEEP = 2  # seconds between polls; matches console's poll_interval default
+
+
+def _usage() -> None:
+    sys.stderr.write(
+        "usage: bb login [--console-url URL] [--label LABEL]\n"
+        "\n"
+        "Options:\n"
+        "  --console-url URL  Console base URL (or BB_CONSOLE_URL env var)\n"
+        "  --label LABEL      Host label shown in the console (default: hostname)\n"
+    )
+
+
+def _flag(argv: list[str], name: str) -> str | None:
+    for i, arg in enumerate(argv):
+        if arg == name and i + 1 < len(argv):
+            return argv[i + 1]
+        if arg.startswith(f"{name}="):
+            return arg[len(name) + 1:]
+    return None
+
+
+def _post(url: str, payload: dict[str, Any]) -> dict[str, Any]:
+    data = json.dumps(payload).encode()
+    req = urllib.request.Request(
+        url, data=data, headers={"Content-Type": "application/json"}
+    )
+    with urllib.request.urlopen(req, timeout=10) as resp:
+        return json.loads(resp.read())
+
+
+def _get(url: str) -> tuple[int, dict[str, Any]]:
+    req = urllib.request.Request(url)
+    try:
+        with urllib.request.urlopen(req, timeout=10) as resp:
+            return resp.status, json.loads(resp.read())
+    except urllib.error.HTTPError as e:
+        return e.code, {}
+
+
+def _save_credentials(
+    console_url: str, host_id: str, access_token: str, refresh_token: str
+) -> Path:
+    path = bot_bottle_root() / "console.json"
+    path.parent.mkdir(parents=True, exist_ok=True)
+    content = (
+        json.dumps(
+            {
+                "url": console_url,
+                "host_id": host_id,
+                "access_token": access_token,
+                "refresh_token": refresh_token,
+            },
+            indent=2,
+        )
+        + "\n"
+    )
+    fd, tmp_path_str = tempfile.mkstemp(dir=path.parent, prefix=".console-")
+    tmp = Path(tmp_path_str)
+    try:
+        tmp.chmod(0o600)
+        with os.fdopen(fd, "w") as f:
+            f.write(content)
+        os.replace(tmp, path)
+    except OSError:
+        try:
+            tmp.unlink()
+        except OSError:
+            pass
+        raise
+    return path
+
+
+def cmd_login(argv: list[str]) -> int:
+    if "--help" in argv or "-h" in argv:
+        _usage()
+        return 0
+
+    console_url = _flag(argv, "--console-url") or os.environ.get(_CONSOLE_URL_ENV)
+    if not console_url:
+        sys.stderr.write(
+            "bb login: --console-url or BB_CONSOLE_URL is required\n"
+        )
+        return 1
+    console_url = console_url.rstrip("/")
+
+    label = _flag(argv, "--label") or socket.gethostname()
+
+    try:
+        resp = _post(f"{console_url}/api/v1/hosts/authorize", {"label": label})
+    except (OSError, ValueError) as exc:
+        sys.stderr.write(f"bb login: failed to start authorization: {exc}\n")
+        return 1
+
+    device_code = resp["device_code"]
+    user_code = resp["user_code"]
+    expires_in = resp.get("expires_in", 300)
+    poll_sleep = max(1, min(int(resp.get("poll_interval", _POLL_SLEEP)), 60))
+
+    sys.stderr.write(
+        f"\nOpen this URL in your browser to authorize this host:\n\n"
+        f"  {console_url}/hosts/authorize?code={user_code}\n\n"
+        f"Waiting for approval"
+    )
+
+    deadline = time.monotonic() + expires_in
+    while time.monotonic() < deadline:
+        sys.stderr.write(".")
+        sys.stderr.flush()
+        time.sleep(poll_sleep)
+
+        try:
+            code, result = _get(
+                f"{console_url}/api/v1/hosts/authorize/{device_code}"
+            )
+        except (OSError, ValueError):
+            continue
+
+        if code == 410:
+            break
+
+        st = result.get("status")
+        if st == "approved":
+            sys.stderr.write("\n\nApproved.\n")
+            path = _save_credentials(
+                console_url,
+                result["host_id"],
+                result["access_token"],
+                result["refresh_token"],
+            )
+            sys.stderr.write(f"Credentials saved to {path}\n")
+            return 0
+        if st == "denied":
+            sys.stderr.write("\n\nDenied by operator.\n")
+            return 1
+
+    sys.stderr.write("\n\nAuthorization timed out.\n")
+    return 1

bot_bottle/cli/commands/resume.py

@@ -0,0 +1,59 @@
+"""resume: re-launch a bottle by its identity.
+
+Reads ~/.bot-bottle/state/<identity>/metadata.json to recover the
+(agent_name, cwd, copy_cwd) the bottle was originally started with,
+then runs the same launch core as `start` — but pinned to the
+recorded identity so the new bottle picks up any per-bottle Dockerfile
+override and transcript snapshot under the same state dir.
+
+Use case: an interrupted or preserved bottle needs to be relaunched;
+the operator runs
+    ./cli.py resume <identity>
+to bring up the replacement from the recorded state.
+"""
+
+from __future__ import annotations
+
+import argparse
+
+from ...backend import BottleSpec
+from ...bottle_state import read_metadata
+from ...log import die
+from ...manifest import ManifestIndex
+from .._common import PROG, USER_CWD
+from .start import _launch_bottle
+
+
+def cmd_resume(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} resume", add_help=True)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument(
+        "identity",
+        help="bottle identity from a prior `start` (see its session-end output)",
+    )
+    args = parser.parse_args(argv)
+
+    metadata = read_metadata(args.identity)
+    if metadata is None:
+        die(
+            f"no state recorded for identity {args.identity!r}; "
+            f"check ~/.bot-bottle/state/ or run `cli.py start` to create a new bottle"
+        )
+
+    manifest = ManifestIndex.resolve(USER_CWD)
+    manifest.require_agent(metadata.agent_name)
+
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=metadata.agent_name,
+        copy_cwd=metadata.copy_cwd,
+        user_cwd=metadata.cwd or USER_CWD,
+        identity=metadata.identity,
+        bottle_names=tuple(metadata.bottle_names),
+    )
+    backend_name = metadata.backend or None
+    return _launch_bottle(
+        spec,
+        dry_run=args.dry_run,
+        backend_name=backend_name,
+    )

bot_bottle/cli/commands/start.py

@@ -0,0 +1,615 @@
+"""start: boot a sandboxed container for a named agent and attach an
+interactive claude-code session. The container is torn down when the
+session ends.
+
+`--headless` selects a non-interactive launch (agent/bottles/label from
+flags, no TUI selectors, no y/N prompt) for orchestrators,
+CI, and webhook dispatch. The agent still execs on the inherited
+stdio/PTY, so an orchestrator that allocates the PTY drives the session.
+
+The launch core is shared with `cli.py resume <identity>` through
+the private orchestrator `_launch_bottle`.
+"""
+
+from __future__ import annotations
+
+import argparse
+import io
+import os
+import shutil
+import sys
+import tempfile
+from pathlib import Path
+from typing import Callable
+
+from ...agent_provider import get_provider, runtime_for
+from ...backend import (
+    Bottle,
+    BottleSpec,
+    enumerate_active_agents,
+    get_bottle_backend,
+)
+from ...backend.docker import util as docker_mod
+from ...backend.docker.bottle_plan import DockerBottlePlan
+from ...bottle_state import (
+    cleanup_state,
+    is_preserved,
+    mark_preserved,
+)
+from ...image_cache import StaleImageError
+from ...log import info, die
+from ...manifest import Manifest, ManifestIndex
+from .._common import PROG, USER_CWD, read_tty_line
+from .. import tui
+
+
+def cmd_start(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} start", add_help=True)
+    parser.add_argument("--dry-run", action="store_true")
+    parser.add_argument("--cwd", action="store_true", help="copy host cwd into the running bottle")
+    parser.add_argument(
+        "--no-cache",
+        action="store_true",
+        help=(
+            "rebuild agent/sidecar images from scratch, bypassing the "
+            "build layer cache. Use when an image looks broken after a "
+            "dependency bump — e.g. an installer's optionalDependencies "
+            "fetch silently no-op'd on a transient failure and got baked "
+            "into a cached layer."
+        ),
+    )
+    parser.add_argument(
+        "--headless",
+        action="store_true",
+        help=(
+            "non-interactive launch: take agent/bottles/label from flags, "
+            "skip all prompts. For orchestrators, CI, and webhooks."
+        ),
+    )
+    parser.add_argument(
+        "--cached-images",
+        action="store_true",
+        help=(
+            "quickstart with existing local agent and sidecar images; "
+            "only valid with --headless"
+        ),
+    )
+    parser.add_argument(
+        "--bottle",
+        action="append",
+        default=None,
+        metavar="NAME",
+        help=(
+            "bottle to compose, repeatable (order = merge order). In "
+            "--headless, defaults to the agent's own bottle when omitted."
+        ),
+    )
+    parser.add_argument(
+        "--label",
+        default=None,
+        help="bottle label / terminal title (--headless default: agent name)",
+    )
+    parser.add_argument(
+        "--color",
+        default=None,
+        help="bottle color, one of the 16 ANSI color names (--headless default: none)",
+    )
+    parser.add_argument(
+        "--prompt",
+        default=None,
+        help="initial task prompt delivered to the agent (required with --headless)",
+    )
+    parser.add_argument(
+        "name",
+        nargs="?",
+        default=None,
+        help="agent name defined in bot-bottle.json (omit to pick interactively)",
+    )
+    args = parser.parse_args(argv)
+    if args.cached_images and not args.headless:
+        die("--cached-images is only supported with --headless")
+
+    dry_run = args.dry_run or os.environ.get("BOT_BOTTLE_DRY_RUN") == "1"
+    if args.no_cache or os.environ.get("BOT_BOTTLE_NO_CACHE") == "1":
+        # Read by build_image() in each backend's util module — set here
+        # so both the interactive and --headless paths pick it up without
+        # threading a no_cache field through every backend's plan dataclass.
+        os.environ["BOT_BOTTLE_NO_CACHE"] = "1"
+
+    manifest = ManifestIndex.resolve(USER_CWD)
+
+    if args.headless:
+        return _start_headless(
+            manifest, args, dry_run=dry_run
+        )
+
+    agent_name: str | None = args.name
+    if agent_name is None:
+        if not manifest.all_agent_names:
+            print(
+                "bot-bottle: no agents defined. "
+                "Add an agent to ~/.bot-bottle/agents/ or ./bot-bottle/agents/ to get started.",
+                file=sys.stderr,
+            )
+            return 1
+        agent_name = tui.filter_select(
+            manifest.all_agent_names,
+            title="Select agent",
+        )
+        if agent_name is None:
+            return 0
+
+    # Bottle multiselect: always show after agent selection so operators
+    # can compose bottles at launch time without editing agent manifests.
+    available_bottles = manifest.all_bottle_names
+    lineage_map = _bottle_lineage(manifest)
+    display_labels = [lineage_map.get(n, n) for n in available_bottles]
+    label_to_name = {lineage_map.get(n, n): n for n in available_bottles}
+    initial_bottle = _peek_agent_bottle(manifest, agent_name)
+    initial_labels = [lineage_map.get(initial_bottle, initial_bottle)] if initial_bottle else []
+    selected_labels = tui.filter_multiselect(
+        display_labels,
+        title="Select bottles",
+        initial=initial_labels,
+    )
+    if selected_labels is None:
+        return 0
+    bottle_names = tuple(label_to_name.get(lbl, lbl) for lbl in selected_labels)
+
+    label, color = tui.name_color_modal(default_label=agent_name)
+    label, color = _resolve_unique_label(label, color)
+
+    image_policy = _select_image_policy()
+    if image_policy is None:
+        return 0
+
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=agent_name,
+        copy_cwd=args.cwd,
+        user_cwd=USER_CWD,
+        label=label,
+        color=color,
+        bottle_names=bottle_names,
+        image_policy=image_policy,
+    )
+    return _launch_bottle(
+        spec,
+        dry_run=dry_run,
+    )
+
+
+# --- Headless launch -----------------------------------------------------
+
+
+def _start_headless(
+    manifest: ManifestIndex,
+    args: argparse.Namespace,
+    *,
+    dry_run: bool,
+) -> int:
+    """Non-interactive launch path for orchestrators / CI / webhooks.
+
+    Resolves agent, bottles, label, and color from flags + manifest
+    defaults instead of the TUI selectors, and auto-confirms the
+    preflight. Otherwise runs the same launch core as the interactive
+    path, so the agent still execs on the inherited stdio/PTY — an
+    orchestrator allocates that PTY and relays it to its
+    desktop/mobile clients."""
+    try:
+        stdin_fd = sys.stdin.fileno()
+    except io.UnsupportedOperation:
+        stdin_fd = -1
+    if not os.isatty(stdin_fd):
+        die(
+            "--headless requires a PTY on stdin; run via:\n"
+            "  script -q /dev/null ./cli.py start ..."
+        )
+
+    agent_name = args.name
+    if not agent_name:
+        die("--headless requires an agent name: ./cli.py start <agent> --headless")
+    manifest.require_agent(agent_name)  # raises ManifestError if unknown
+
+    prompt = args.prompt
+    if not prompt:
+        die(
+            "--headless requires --prompt: "
+            "./cli.py start <agent> --headless --prompt 'Do the thing'"
+        )
+
+    if args.bottle:
+        bottle_names: tuple[str, ...] = tuple(args.bottle)
+    else:
+        default_bottle = _peek_agent_bottle(manifest, agent_name)
+        if not default_bottle:
+            die(
+                f"--headless: agent '{agent_name}' has no default bottle; "
+                f"pass one or more --bottle NAME"
+            )
+        bottle_names = (default_bottle,)
+
+    label = _uniquify_label_headless(args.label or agent_name)
+
+    spec = BottleSpec(
+        manifest=manifest,
+        agent_name=agent_name,
+        copy_cwd=args.cwd,
+        user_cwd=USER_CWD,
+        label=label,
+        color=args.color or "",
+        bottle_names=bottle_names,
+        headless=True,
+        image_policy="cached" if args.cached_images else "fresh",
+    )
+    return _launch_bottle(
+        spec,
+        dry_run=dry_run,
+        assume_yes=True,
+        headless_prompt_text=prompt,
+    )
+
+
+def _uniquify_label_headless(label: str) -> str:
+    """Non-interactive analog of `_resolve_unique_label`: if the label's
+    slug collides with a running bottle, append -2, -3, … until free,
+    logging the chosen label. Orchestrators fire-and-forget many bottles,
+    so silently picking a free name beats erroring on every collision."""
+    active_slugs = {a.slug for a in enumerate_active_agents()}
+    if docker_mod.slugify(label) not in active_slugs:
+        return label
+    n = 2
+    while docker_mod.slugify(f"{label}-{n}") in active_slugs:
+        n += 1
+    chosen = f"{label}-{n}"
+    info(f"label '{label}' already in use; using '{chosen}'")
+    return chosen
+
+
+# --- Launch helpers ------------------------------------------------------
+
+
+def prepare_with_preflight(
+    spec: BottleSpec,
+    *,
+    stage_dir: Path,
+    render_preflight: Callable[[DockerBottlePlan, str], 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.
+
+    `backend_name` selects which backend prepares the plan
+    (`None` → `$BOT_BOTTLE_BACKEND` → host auto-selection).
+
+    When `spec.headless` is True the docker-fallback prompt is suppressed:
+    auto-selection dies with an actionable message rather than blocking
+    on a TTY read (which would hang CI, webhook dispatch, and orchestrators).
+
+    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_name, prompt=not spec.headless)
+    plan = backend.prepare(spec, stage_dir=stage_dir)
+    identity = _identity_from_plan(plan)
+
+    render_preflight(plan, backend.name)
+
+    if dry_run:
+        info("dry-run requested; not starting container.")
+        return None, identity
+    if not prompt_yes():
+        info("aborted by user")
+        return None, identity
+    return plan, identity
+
+
+def attach_agent(
+    bottle: Bottle, *, resume: bool = False,
+    agent_provider_template: str = "claude",
+    startup_args: tuple[str, ...] = (),
+) -> int:
+    """Run the selected provider CLI inside `bottle` as an
+    interactive session. Blocks until the session ends; returns the
+    agent process's exit code.
+
+    `resume=True` adds `--continue` so claude picks up its most
+    recent session non-interactively (no session-picker prompt).
+    First-attach paths (`./cli.py start`) leave it False.
+
+    Used as the inner step of `./cli.py start`."""
+    runtime = runtime_for(agent_provider_template)
+    info(
+        f"attaching interactive {agent_provider_template} session "
+        "(Ctrl-D or 'exit' to leave; container will be removed)"
+    )
+    agent_args = list(runtime.bypass_args)
+    agent_args.extend(startup_args)
+    if resume:
+        agent_args.extend(runtime.resume_args)
+    return bottle.exec_agent(agent_args, tty=True)
+
+
+def capture_claude_session_state(identity: str, exit_code: int) -> None:
+    """Inside the launch context, while the container is still
+    alive: snapshot the transcript and mark for preservation if
+    claude crashed."""
+    # FIXME: this captures Claude-specific session state. A follow-up
+    # spike should explore freezing provider-neutral container state
+    # instead of relying on each agent's transcript layout.
+    if not identity:
+        return
+    # snapshot_transcript(identity)
+    if exit_code != 0:
+        mark_preserved(identity)
+
+
+def settle_state(identity: str) -> None:
+    """Post-teardown housekeeping: print the resume hint if the
+    state was preserved, otherwise reap the per-bottle state dir."""
+    if not identity:
+        return
+    if is_preserved(identity):
+        info(f"to resume this bottle: ./cli.py resume {identity}")
+        return
+    cleanup_state(identity)
+
+
+def _identity_from_plan(plan: object) -> str:
+    """Backend-specific: the docker plan exposes the identity as
+    `.slug`. Other backends in the future would expose their own
+    identity attribute; for now we duck-type to keep this layer
+    backend-agnostic."""
+    return getattr(plan, "slug", "")
+
+
+def _peek_agent_bottle(manifest: ManifestIndex, agent_name: str) -> str:
+    """Return the `bottle:` value from the named agent's frontmatter without
+    fully parsing the agent file, or "" when absent or unreadable.
+
+    Used to pre-populate the bottle multiselect with the agent's default
+    bottle so operators who haven't removed `bottle:` from their manifests
+    don't need to re-select it every time."""
+    if manifest.home_md is None:
+        # Eager mode (from_json_obj): agent is pre-parsed.
+        if agent_name in manifest.agents:
+            return manifest.agents[agent_name].bottle
+        return ""
+
+    from ...manifest.loader import scan_agent_names
+    from ...yaml_subset import YamlSubsetError, parse_frontmatter
+
+    home_agents = scan_agent_names(manifest.home_md / "agents")
+    cwd_agents: dict[str, Path] = {}
+    if manifest.cwd_md is not None:
+        cwd_agents = scan_agent_names(manifest.cwd_md / "agents")
+    merged = {**home_agents, **cwd_agents}
+    path = merged.get(agent_name)
+    if path is None:
+        return ""
+    try:
+        fm, _ = parse_frontmatter(path.read_text())
+        bottle = fm.get("bottle", "")
+        return str(bottle) if isinstance(bottle, str) else ""
+    except (OSError, YamlSubsetError):
+        return ""
+
+
+def _resolve_unique_label(label: str, color: str) -> tuple[str, str]:
+    """Re-prompt with a disclaimer until the label's slug is not already
+    in use among running bottles.  Passes through unchanged when no
+    collision is found on the first check."""
+    while True:
+        slug_candidate = docker_mod.slugify(label)
+        active_slugs = {a.slug for a in enumerate_active_agents()}
+        if slug_candidate not in active_slugs:
+            return label, color
+        label, color = tui.name_color_modal(
+            default_label=label,
+            disclaimer=f'"{label}" is already in use',
+        )
+
+
+def _text_prompt_yes() -> bool:
+    """Default `prompt_yes` for CLI use: reads y/N from the
+    controlling tty via stderr prompt + tty-line read."""
+    sys.stderr.write("bot-bottle: launch this agent? [y/N] ")
+    sys.stderr.flush()
+    reply = read_tty_line()
+    return reply in ("y", "Y", "yes", "YES")
+
+
+def _select_image_policy() -> str | None:
+    return tui.filter_select(
+        ["fresh", "cached"],
+        title="Select image startup mode",
+    )
+
+
+def _text_render_preflight():
+    def _render(plan: DockerBottlePlan, backend_name: str) -> None:
+        print(file=sys.stderr)
+        print(f"backend: {backend_name}", file=sys.stderr)
+        print(_manifest_to_yaml(plan.manifest), file=sys.stderr)
+    return _render
+
+
+def _bottle_lineage(manifest: ManifestIndex) -> dict[str, str]:
+    """Return {bottle_name: lineage_label} for bottles that have an extends chain.
+
+    Bottles without a parent are omitted (the caller falls back to the bare name).
+    Labels show the chain root-first: e.g. 'dev -> bot-bottle-dev -> claude-dev'."""
+    if manifest.home_md is None:
+        return {}
+    bottles_dir = manifest.home_md / "bottles"
+    if not bottles_dir.is_dir():
+        return {}
+
+    from ...yaml_subset import YamlSubsetError, parse_frontmatter
+
+    extends_of: dict[str, str] = {}
+    for path in bottles_dir.glob("*.md"):
+        try:
+            fm, _ = parse_frontmatter(path.read_text())
+            parent = fm.get("extends", "")
+            if isinstance(parent, str) and parent:
+                extends_of[path.stem] = parent
+        except (OSError, YamlSubsetError):
+            pass
+
+    labels: dict[str, str] = {}
+    for name in extends_of:
+        chain = [name]
+        seen = {name}
+        cur = name
+        while cur in extends_of:
+            par = extends_of[cur]
+            if par in seen:
+                break
+            chain.append(par)
+            seen.add(par)
+            cur = par
+        labels[name] = " -> ".join(reversed(chain))
+
+    return labels
+
+
+def _manifest_to_yaml(manifest: Manifest) -> str:
+    """Serialize the resolved Manifest to a YAML string for preflight display."""
+    lines: list[str] = []
+
+    agent = manifest.agent
+    lines.append("agent:")
+    if agent.skills:
+        lines.append("  skills:")
+        for s in agent.skills:
+            lines.append(f"    - {s}")
+    if not agent.git_user.is_empty():
+        lines.append("  git-gate:")
+        lines.append("    user:")
+        if agent.git_user.name:
+            lines.append(f"      name: {agent.git_user.name}")
+        if agent.git_user.email:
+            lines.append(f"      email: {agent.git_user.email}")
+
+    bottle = manifest.bottle
+    lines.append("bottle:")
+
+    if bottle.agent_provider.template != "claude" or bottle.agent_provider.dockerfile:
+        lines.append("  agent_provider:")
+        lines.append(f"    template: {bottle.agent_provider.template}")
+        if bottle.agent_provider.dockerfile:
+            lines.append(f"    dockerfile: {bottle.agent_provider.dockerfile}")
+
+    if bottle.env:
+        lines.append("  env:")
+        for k, v in sorted(bottle.env.items()):
+            lines.append(f"    {k}: {v}")
+
+    has_git_gate = not bottle.git_user.is_empty() or bottle.git
+    if has_git_gate:
+        lines.append("  git-gate:")
+        if not bottle.git_user.is_empty():
+            lines.append("    user:")
+            if bottle.git_user.name:
+                lines.append(f"      name: {bottle.git_user.name}")
+            if bottle.git_user.email:
+                lines.append(f"      email: {bottle.git_user.email}")
+        if bottle.git:
+            lines.append("    repos:")
+            for entry in bottle.git:
+                lines.append(f"      {entry.Name}:")
+                lines.append(f"        url: {entry.Upstream}")
+
+    if bottle.egress.routes:
+        lines.append("  egress:")
+        lines.append("    routes:")
+        for r in bottle.egress.routes:
+            lines.append(f"      - host: {r.Host}")
+            if r.AuthScheme:
+                lines.append(f"        auth:")
+                lines.append(f"          scheme: {r.AuthScheme}")
+
+    lines.append(f"  supervise: {'true' if bottle.supervise else 'false'}")
+    if bottle.nested_containers:
+        lines.append("  nested_containers: true")
+
+    return "\n".join(lines)
+
+
+def _launch_bottle(
+    spec: BottleSpec,
+    *,
+    dry_run: bool,
+    backend_name: str | None = None,
+    assume_yes: bool = False,
+    headless_prompt_text: str = "",
+) -> int:
+    """Shared launch core for `start` and `resume`. Builds the plan,
+    prints / dry-runs / prompts as appropriate, brings the bottle up,
+    attaches claude, and prints the resume hint on session end.
+
+    `assume_yes` skips the interactive y/N confirmation (headless /
+    orchestrator launches), where there is no human at the prompt.
+
+    `headless_prompt_text` is passed to the provider's `headless_prompt`
+    method and the resulting args are appended to startup_args so the
+    agent receives the initial task without interactive input."""
+    stage_dir = Path(tempfile.mkdtemp(prefix="bot-bottle-stage."))
+    identity = ""
+    try:
+        plan, identity = prepare_with_preflight(
+            spec,
+            stage_dir=stage_dir,
+            render_preflight=_text_render_preflight(),
+            prompt_yes=(lambda: True) if assume_yes else _text_prompt_yes,
+            dry_run=dry_run,
+            backend_name=backend_name,
+        )
+        if plan is None:
+            return 0
+
+        backend = get_bottle_backend(backend_name)
+        try:
+            backend.prelaunch_checks(plan)
+        except StaleImageError as exc:
+            if assume_yes:
+                die(str(exc))
+            sys.stderr.write(f"bot-bottle: {exc}\nLaunch anyway? [y/N] ")
+            sys.stderr.flush()
+            if read_tty_line() not in ("y", "Y", "yes", "YES"):
+                return 0
+        with backend.launch(plan) as bottle:
+            agent_provider_template = getattr(plan, "agent_provider_template", "claude")
+            extra_args: tuple[str, ...] = ()
+            if headless_prompt_text:
+                extra_args = tuple(
+                    get_provider(agent_provider_template).headless_prompt(
+                        headless_prompt_text
+                    )
+                )
+            exit_code = attach_agent(
+                bottle,
+                agent_provider_template=agent_provider_template,
+                startup_args=plan.agent_provision.startup_args + extra_args,
+            )
+            info(
+                f"session ended (exit {exit_code}); "
+                f"container {bottle.name} will be removed"
+            )
+            if agent_provider_template == "claude":
+                capture_claude_session_state(identity, exit_code)
+        return 0
+    finally:
+        # PRD 0018 chunk 2: prepare now writes the bottle's bind-mount
+        # sources under state/<slug>/. If we never reached the
+        # launch context (dry-run, preflight-N, prepare exception), or
+        # we did but nothing requested preservation, reap them along
+        # with everything else. `settle_state` subsumes the prior
+        # post-launch settlement and the new pre-launch cleanup.
+        settle_state(identity)
+        shutil.rmtree(stage_dir, ignore_errors=True)

bot_bottle/cli/commands/supervise.py

@@ -0,0 +1,546 @@
+"""supervise: list pending supervise proposals across all bottles and
+act on them (approve / modify / reject).
+
+Curses-based TUI; modify-then-approve shells out to $EDITOR. The
+Egress proposals are queued for operator review as full routes.yaml
+updates.
+"""
+
+from __future__ import annotations
+
+import argparse
+import curses
+import os
+import subprocess
+import sys
+import tempfile
+import traceback
+from dataclasses import dataclass
+from datetime import datetime, timezone
+from pathlib import Path
+
+from ...paths import bot_bottle_root
+from ...log import Die, error, info
+from ...orchestrator.client import (
+    OrchestratorClient,
+    OrchestratorClientError,
+    discover_orchestrator_url,
+)
+
+from ...supervisor.types import (
+    Proposal,
+    TOOL_EGRESS_ALLOW,
+    TOOL_EGRESS_BLOCK,
+    TOOL_GITLEAKS_ALLOW,
+    TOOL_EGRESS_TOKEN_ALLOW,
+)
+from .._common import PROG
+
+
+_REFRESH_INTERVAL_MS = 1000
+
+# Proposal tools whose payload is a read-only report, not a file the operator
+# edits: modify is unavailable and approval requires a recorded reason for the
+# audit trail.
+_REPORT_ONLY_TOOLS: tuple[str, ...] = (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW)
+
+
+@dataclass(frozen=True)
+class QueuedProposal:
+    """A pending proposal from the supervise queue.
+
+    `label` is the operator-facing bottle name (the human slug the
+    orchestrator resolved from the registry); `proposal.bottle_slug` is the
+    opaque bottle_id every operator action is keyed by. Display uses `label`;
+    respond calls use `proposal.bottle_slug`."""
+
+    proposal: Proposal
+    label: str = ""
+
+
+# A failed operator action (orchestrator unreachable, bottle torn down,
+# 409) is caught by the TUI key handlers and surfaced in the status line so
+# the proposal stays pending rather than crashing curses.
+ApplyError = (OrchestratorClientError,)
+
+
+# The one per-host orchestrator, discovered lazily on first use. Every
+# operator action — list, approve, reject — goes through its HTTP control
+# plane (the orchestrator owns the single DB + live policy); there is no
+# direct-DB path and no backend branching here.
+_client_instance: OrchestratorClient | None = None
+
+
+def _resolve_orchestrator_url() -> str:
+    """URL of the running orchestrator control plane, starting one on demand.
+
+    Supervise is often the first thing an operator runs — before any bottle
+    has booted the control plane. So when discovery finds nothing, bring up
+    the selected backend's orchestrator + gateway (idempotent) rather than
+    failing with "launch a bottle first"."""
+    try:
+        return discover_orchestrator_url()
+    except OrchestratorClientError:
+        from ...backend import get_bottle_backend
+        backend = get_bottle_backend()
+        info(f"no orchestrator control plane running; starting one ({backend.name})…")
+        return backend.ensure_orchestrator()
+
+
+def _client() -> OrchestratorClient:
+    global _client_instance  # noqa: PLW0603 — CLI-session singleton
+    if _client_instance is None:
+        _client_instance = OrchestratorClient(_resolve_orchestrator_url())
+    return _client_instance
+
+
+def discover_pending() -> list[QueuedProposal]:
+    """Collect pending proposals across bottles from the orchestrator."""
+    out = [
+        QueuedProposal(
+            proposal=Proposal.from_dict(d),
+            label=str(d.get("bottle_label") or d.get("bottle_slug") or ""),
+        )
+        for d in _client().supervise_pending()
+    ]
+    out.sort(key=lambda q: q.proposal.arrival_timestamp)
+    return out
+
+
+def _approval_status(qp: QueuedProposal, verb: str) -> str:
+    """Status-line text after a successful approval."""
+    base = f"{verb} {qp.proposal.tool} for [{qp.label}]"
+    return f"{base}; resume: ./cli.py resume {qp.label}"
+
+
+def _detail_lines(
+    qp: QueuedProposal,
+    *,
+    green_attr: int = 0,
+) -> list[tuple[str, int]]:
+    """Return the detail-view body as (text, curses-attr) tuples."""
+    p = qp.proposal
+    out: list[tuple[str, int]] = [
+        (f"bottle: {qp.label}", 0),
+        (f"tool: {p.tool}", 0),
+        (f"id: {p.id}", 0),
+        (f"arrived: {p.arrival_timestamp}", 0),
+        ("", 0),
+        ("justification:", 0),
+    ]
+    out.extend(("  " + line, 0) for line in p.justification.splitlines() or [""])
+    out.extend([
+        ("", 0),
+        ("proposed file:", 0),
+    ])
+    out.extend((line, 0) for line in p.proposed_file.splitlines() or [""])
+    return out
+
+
+def _suffix_for_tool(tool: str) -> str:
+    if tool in (TOOL_EGRESS_ALLOW, TOOL_EGRESS_BLOCK):
+        return ".yaml"
+    if tool in (TOOL_GITLEAKS_ALLOW, TOOL_EGRESS_TOKEN_ALLOW):
+        return ".txt"
+    return ".txt"
+
+
+# --- Operator actions ------------------------------------------------------
+
+
+def approve(
+    qp: QueuedProposal,
+    *,
+    notes: str = "",
+    final_file: str | None = None,
+) -> None:
+    """Approve (or, with `final_file`, modify-then-approve) via the
+    orchestrator: it applies the route change to the bottle's live policy,
+    writes the response that unblocks the agent, and audits it — one atomic
+    server-side op. Raises `OrchestratorClientError` on failure."""
+    _client().supervise_respond(
+        qp.proposal.id,
+        bottle_slug=qp.proposal.bottle_slug,
+        decision="modify" if final_file is not None else "approve",
+        notes=notes,
+        final_file=final_file,
+    )
+
+
+def reject(qp: QueuedProposal, *, reason: str) -> None:
+    """Reject via the orchestrator (writes the response + audit)."""
+    _client().supervise_respond(
+        qp.proposal.id,
+        bottle_slug=qp.proposal.bottle_slug,
+        decision="reject",
+        notes=reason,
+    )
+
+
+def _approve_from_tui(
+    stdscr: "curses._CursesWindow",  # type: ignore
+    qp: QueuedProposal,
+    *,
+    final_file: str | None = None,
+    notes: str = "",
+) -> str:
+    """Approve from curses, prompting for any tool-specific audit note."""
+    if qp.proposal.tool in _REPORT_ONLY_TOOLS and final_file is None:
+        notes = _prompt(stdscr, "allow reason (false positive / legitimately needed): ")
+        if not notes:
+            return "approve aborted (empty reason)"
+    approve(qp, final_file=final_file, notes=notes)
+    verb = "modified+approved" if final_file is not None else "approved"
+    return _approval_status(qp, verb)
+
+
+# --- $EDITOR integration --------------------------------------------------
+
+
+def edit_in_editor(content: str, *, suffix: str = ".tmp") -> str | None:
+    """Open `content` in $EDITOR and return edited content, if changed."""
+    editor = os.environ.get("EDITOR", "vim")
+    with tempfile.NamedTemporaryFile(
+        mode="w", suffix=suffix, delete=False, prefix="supervise-modify.",
+    ) as f:
+        f.write(content)
+        path = f.name
+    try:
+        subprocess.run([editor, path], check=False)
+        with open(path, encoding="utf-8") as f:
+            edited = f.read()
+        return edited if edited != content else None
+    finally:
+        try:
+            os.unlink(path)
+        except OSError:
+            pass
+
+
+# --- TUI -------------------------------------------------------------------
+
+
+def cmd_supervise(argv: list[str]) -> int:
+    parser = argparse.ArgumentParser(prog=f"{PROG} supervise", add_help=True)
+    parser.add_argument(
+        "--once", action="store_true",
+        help="list pending proposals once and exit (no TUI)",
+    )
+    args = parser.parse_args(argv)
+
+    # Establish the orchestrator connection up front so a missing control
+    # plane is a clean one-line error, not a curses crash mid-loop. This also
+    # starts the orchestrator on demand when none is running (see `_client`).
+    try:
+        _client()
+    except OrchestratorClientError as e:
+        error(str(e))
+        return 1
+    except Die as e:
+        # Backend has no orchestrator to start (e.g. macos-container).
+        if e.message:
+            error(e.message)
+        return e.code if isinstance(e.code, int) else 1
+
+    if args.once:
+        return _list_once()
+    try:
+        curses.wrapper(_main_loop)
+    except KeyboardInterrupt:
+        return 130
+    except Die as e:
+        if e.message:
+            error(e.message)
+        else:
+            error("supervise exited on a fatal error (no detail captured).")
+        return e.code if isinstance(e.code, int) else 1
+    except Exception as e:  # noqa: W0718 — catch supervise crash for logging
+        log_path = _write_crash_log(e)
+        error(
+            f"supervise crashed: {type(e).__name__}: {e}",
+            context={"error_type": type(e).__name__, "crash_log": str(log_path)},
+        )
+        error(f"full traceback written to {log_path}")
+        return 1
+    return 0
+
+
+def _write_crash_log(exc: BaseException) -> Path:
+    """Persist `exc`'s traceback to a stable file under ~/.bot-bottle/."""
+    stamp = datetime.now(timezone.utc).strftime("%Y-%m-%dT%H:%M:%SZ")
+    body = "".join(
+        traceback.format_exception(type(exc), exc, exc.__traceback__)
+    )
+    entry = f"=== supervise crash {stamp} ===\n{body}\n"
+    try:
+        log_dir = bot_bottle_root() / "logs"
+        log_dir.mkdir(parents=True, exist_ok=True)
+        path = log_dir / "supervise-crash.log"
+        with path.open("a", encoding="utf-8") as fh:
+            fh.write(entry)
+        return path
+    except OSError:
+        fd, tmp = tempfile.mkstemp(
+            prefix="bot-bottle-supervise-crash-", suffix=".log",
+        )
+        with os.fdopen(fd, "w", encoding="utf-8") as fh:
+            fh.write(entry)
+        return Path(tmp)
+
+
+def _list_once() -> int:
+    pending = discover_pending()
+    if not pending:
+        info("no pending proposals")
+        return 0
+    for qp in pending:
+        sys.stdout.write(
+            f"{qp.proposal.arrival_timestamp}  "
+            f"[{qp.label}]  "
+            f"{qp.proposal.tool}  "
+            f"{qp.proposal.id}\n"
+        )
+        sys.stdout.write(f"    {qp.proposal.justification}\n")
+    return 0
+
+
+def _try_init_green() -> int:  # pragma: no cover
+    """Initialise a green color pair and return its attr, or 0."""
+    try:
+        curses.start_color()
+        curses.use_default_colors()
+        curses.init_pair(1, curses.COLOR_GREEN, -1)
+        return curses.color_pair(1)
+    except curses.error:
+        return 0
+
+
+def _main_loop(stdscr: "curses._CursesWindow") -> None:  # type: ignore  # pragma: no cover
+    curses.curs_set(0)
+    stdscr.timeout(_REFRESH_INTERVAL_MS)
+    green_attr = _try_init_green()
+    selected = 0
+    status_line = ""
+    seen_ids: set[str] = set()
+
+    while True:
+        pending = discover_pending()
+        if selected >= len(pending):
+            selected = max(0, len(pending) - 1)
+
+        live_ids = {qp.proposal.id for qp in pending}
+        newly_arrived = live_ids - seen_ids
+        if seen_ids and newly_arrived:
+            try:
+                curses.beep()
+            except curses.error:
+                pass
+            for i, qp in enumerate(pending):
+                if qp.proposal.id in newly_arrived:
+                    selected = i
+                    break
+        seen_ids = live_ids
+
+        _render(
+            stdscr, pending, selected, status_line,
+            green_attr=green_attr,
+        )
+
+        try:
+            key = stdscr.getch()
+        except KeyboardInterrupt:
+            return
+
+        if key == -1:
+            continue
+
+        status_line = ""
+
+        if key in (ord("q"), 27):
+            return
+
+        if not pending:
+            continue
+        qp = pending[selected]
+
+        if key in (curses.KEY_DOWN, ord("j")):
+            selected = min(selected + 1, len(pending) - 1)
+        elif key in (curses.KEY_UP, ord("k")):
+            selected = max(selected - 1, 0)
+        elif key in (curses.KEY_ENTER, 10, 13):
+            _detail_view(stdscr, qp, green_attr=green_attr)
+        elif key == ord("a"):
+            try:
+                status_line = _approve_from_tui(stdscr, qp)
+            except ApplyError as e:
+                status_line = f"apply failed: {e}"
+        elif key == ord("m"):
+            if qp.proposal.tool in _REPORT_ONLY_TOOLS:
+                status_line = f"modify unavailable for {qp.proposal.tool}"
+                continue
+            edited = _modify(stdscr, qp)
+            if edited is None:
+                status_line = "modify aborted (no change)"
+            else:
+                try:
+                    status_line = _approve_from_tui(
+                        stdscr, qp, final_file=edited,
+                        notes="operator modified before approving",
+                    )
+                except ApplyError as e:
+                    status_line = f"apply failed: {e}"
+        elif key == ord("r"):
+            reason = _prompt(stdscr, "reject reason: ")
+            if reason:
+                reject(qp, reason=reason)
+                status_line = f"rejected {qp.proposal.tool} for [{qp.label}]"
+            else:
+                status_line = "reject aborted (empty reason)"
+
+
+def _render(
+    stdscr: "curses._CursesWindow",  # type: ignore
+    pending: list[QueuedProposal],
+    selected: int,
+    status_line: str,
+    *,
+    green_attr: int = 0,  # noqa: F841 — unused, but required by interface
+) -> None:  # pragma: no cover
+    stdscr.erase()
+    h, w = stdscr.getmaxyx()
+    header = f"bot-bottle supervise  ({len(pending)} pending)"
+    stdscr.addnstr(0, 0, header, w - 1, curses.A_BOLD)
+    stdscr.hline(1, 0, curses.ACS_HLINE, w)
+
+    row = 2
+    if not pending:
+        stdscr.addnstr(
+            row, 2,
+            "no pending proposals; agents will queue here when they call a "
+            "supervise tool",
+            w - 4,
+        )
+    else:
+        for i, qp in enumerate(pending):
+            if row >= h - 3:
+                break
+            p = qp.proposal
+            ts_short = (
+                p.arrival_timestamp.split("T", 1)[1][:8]
+                if "T" in p.arrival_timestamp else p.arrival_timestamp
+            )
+            cursor = "> " if i == selected else "  "
+            line = (
+                f"{cursor}{ts_short}  "
+                f"[{qp.label}]  {p.tool:<18}  {p.id[:8]}"
+            )
+            attr = curses.A_REVERSE if i == selected else curses.A_NORMAL
+            stdscr.addnstr(row, 0, line, w - 1, attr)
+            row += 1
+            if row >= h - 3:
+                break
+            if p.justification:
+                stdscr.addnstr(row, 4, p.justification[: max(0, w - 5)], w - 5)
+                row += 1
+
+    footer = "[j/k] move  [Enter] view  [a] approve  [m] modify  [r] reject  [q] quit"
+    stdscr.hline(h - 2, 0, curses.ACS_HLINE, w)
+    stdscr.addnstr(h - 1, 0, footer, w - 1, curses.A_DIM)
+    if status_line:
+        stdscr.addnstr(h - 3, 0, status_line, w - 1, curses.A_BOLD)
+    stdscr.refresh()
+
+
+def _detail_view(
+    stdscr: "curses._CursesWindow",  # type: ignore
+    qp: QueuedProposal,
+    *,
+    green_attr: int = 0,
+) -> None:  # pragma: no cover
+    """Render the full proposal. Scrollable. Press q to return."""
+    lines = _detail_lines(qp, green_attr=green_attr)
+    offset = 0
+    while True:
+        stdscr.erase()
+        h, w = stdscr.getmaxyx()
+        for i, (text, attr) in enumerate(lines[offset:offset + h - 1]):
+            stdscr.addnstr(i, 0, text, w - 1, attr)
+        stdscr.addnstr(
+            h - 1, 0,
+            "[j/k] scroll  [g/G] top/bottom  [a] approve  [m] modify  [r] reject  [q] back",
+            w - 1, curses.A_DIM,
+        )
+        stdscr.refresh()
+        key = stdscr.getch()
+        if key in (ord("q"), 27):
+            return
+        if key in (curses.KEY_DOWN, ord("j")):
+            offset = min(offset + 1, max(0, len(lines) - 1))
+        elif key in (curses.KEY_UP, ord("k")):
+            offset = max(offset - 1, 0)
+        elif key == ord("g"):
+            offset = 0
+        elif key == ord("G"):
+            offset = max(0, len(lines) - 1)
+        elif key == ord("a"):
+            try:
+                _approve_from_tui(stdscr, qp)
+            except ApplyError:
+                pass
+            return
+        elif key == ord("m"):
+            if qp.proposal.tool in _REPORT_ONLY_TOOLS:
+                return
+            edited = _modify(stdscr, qp)
+            if edited is not None:
+                try:
+                    _approve_from_tui(
+                        stdscr, qp, final_file=edited,
+                        notes="operator modified before approving",
+                    )
+                except ApplyError:
+                    pass
+            return
+        elif key == ord("r"):
+            reason = _prompt(stdscr, "reject reason: ")
+            if reason:
+                reject(qp, reason=reason)
+            return
+
+
+def _modify(stdscr: "curses._CursesWindow", qp: QueuedProposal) -> str | None:  # type: ignore  # pragma: no cover
+    """Suspend curses, open $EDITOR on the proposed file, return edited content."""
+    suffix = _suffix_for_tool(qp.proposal.tool)
+    curses.endwin()
+    try:
+        edited = edit_in_editor(qp.proposal.proposed_file, suffix=suffix)
+    finally:
+        stdscr.refresh()
+    return edited
+
+
+def _prompt(stdscr: "curses._CursesWindow", label: str) -> str:  # type: ignore  # pragma: no cover
+    """One-line input at the bottom of the screen."""
+    curses.curs_set(1)
+    h, _ = stdscr.getmaxyx()
+    stdscr.move(h - 2, 0)
+    stdscr.clrtoeol()
+    stdscr.addstr(h - 2, 0, label)
+    stdscr.refresh()
+    curses.echo()
+    try:
+        raw = stdscr.getstr(h - 2, len(label), 200)
+    finally:
+        curses.noecho()
+        curses.curs_set(0)
+    return raw.decode("utf-8", errors="replace").strip()
+
+
+__all__ = [
+    "QueuedProposal",
+    "approve",
+    "cmd_supervise",
+    "discover_pending",
+    "edit_in_editor",
+    "reject",
+]