#!/usr/bin/env -S uv run --script
# /// script
# requires-python = ">=3.14"
# dependencies = [
#   "typer>=0.12,<1",
#   "questionary>=2.0,<3",
#   "rich>=13,<15",
# ]
# ///
#
# agent-container — interactive wizard + CLI for agent-container containers.
#
# ============================================================================
# This script is the SINGLE SOURCE OF TRUTH for the agent-container on-disk contract.
# Everything below — container naming (agent-container-<name>), the port hash
# (2200 + sum-of-char-codes mod 100), the per-name state files
# ($XDG_STATE_HOME/agent-container/<name>.port), the env-file resolution order
# (./.env -> ~/.config/agent-container/<name>.env -> ~/.config/agent-container/.env), and
# the hosts.conf KEY=VALUE format (FOO_HOST / FOO_PORT) — is defined here and
# only here. The shell completions read the same state files directly; keep
# them in step with the constants in this file.
# ============================================================================
#
# hosts.conf handling: this tool parses hosts.conf line-by-line and NEVER
# executes/sources it. Quoting and unquoted trailing `# comment`s are handled
# like bash `source`, but values containing `$` or backticks are taken
# literally (no expansion), with a one-time warning.

import atexit
import contextlib
import fcntl
import hashlib
import json
import os
import re
import shutil
import signal
import socket
import subprocess
import sys
import tempfile
import termios
import time
import urllib.error
import urllib.request
from collections.abc import Callable
from pathlib import Path
from typing import NoReturn

import questionary
import typer
from rich.console import Console
from rich.table import Table

# --- constants ---------------------------------------------------------------

IMAGE_NAME = "localhost/agent-container:latest"
CONTAINER_PREFIX = "agent-container-"
PORT_BASE = 2200
PORT_RANGE = 100  # 2200..2299

STATE_DIR = (
    Path(os.environ.get("XDG_STATE_HOME") or Path.home() / ".local/state") / "agent-container"
)
CONFIG_DIR = Path(os.environ.get("XDG_CONFIG_HOME") or Path.home() / ".config") / "agent-container"
HOSTS_CONF = CONFIG_DIR / "hosts.conf"
# Feature 001: the host registry (hosts.json) supersedes the flat hosts.conf
# address book. DEFAULT_HOST is the implicit local host name; per-host runtime
# state lives under STATE_DIR/<host>/. See specs/001-multi-host-deployment/.
HOSTS_JSON = CONFIG_DIR / "hosts.json"
REGISTRY_VERSION = 1
DEFAULT_HOST = "local"


def _is_repo_checkout(base: Path) -> bool:
    """Best-effort test that `base` is an agent-container checkout.

    Keyed on `completions/agent-container.bash` (plus a Dockerfile for the build
    context): a repo-specific sentinel rather than the generic `Dockerfile` +
    `completions/` dir pair, which is common to unrelated trees. Requiring the
    exact file _completion_script reads (bin/agent-container: REPO_ROOT/completions/
    agent-container.<shell>) also makes the marker and that consumed path the same
    invariant — a recognized checkout can never then serve stale package data.
    This is a heuristic, not a guaranteed false-positive-free identifier.
    """
    return (base / "Dockerfile").is_file() and (
        base / "completions" / "agent-container.bash"
    ).is_file()


def _find_repo_root() -> Path | None:
    """Locate the repo checkout in a way that survives a non-editable install.

    A wheel copies this module into site-packages, so Path(__file__) no longer
    points inside the repo. Resolution order:
      1. AGENT_CONTAINER_REPO (explicit operator override): trusted only when it
         satisfies the checkout marker (see _is_repo_checkout); a wrong/typo'd
         path yields None rather than a bogus root (build then dies actionably).
      2. Walk up from this file, then from cwd, looking for a dir that matches
         the marker.
    Returns None when no checkout is reachable — client subcommands still work
    (completions falls back to bundled package data; build fails actionably).
    Note: this cannot die here — it runs at import, before Fatal/die exist.
    """
    env = os.environ.get("AGENT_CONTAINER_REPO")
    if env:
        base = Path(env).expanduser().resolve()
        return base if _is_repo_checkout(base) else None
    here = Path(__file__).resolve()
    for base in here.parents:
        if _is_repo_checkout(base):
            return base
    for base in (Path.cwd(), *Path.cwd().parents):
        if _is_repo_checkout(base):
            return base
    return None


REPO_ROOT = _find_repo_root()

NAME_RE = re.compile(r"[a-z0-9][a-z0-9_-]*")
# ssh argv injection guard: a user/host that begins with '-' would be parsed
# by ssh as an option (e.g. -oProxyCommand=...). Conservative charsets that
# can never start with '-'.
SSH_USER_RE = re.compile(r"[A-Za-z0-9_][A-Za-z0-9._-]*")
SSH_HOST_RE = re.compile(r"[A-Za-z0-9_\[][A-Za-z0-9.:_\]-]*")
# tmux window name: embedded in the ssh remote command (attach --window), so it
# is charset-validated. Matches the entrypoint / agent-container window charset.
WINDOW_RE = re.compile(r"[A-Za-z0-9._-]+")

console = Console()


def eprint(msg: str) -> None:
    # Plain stderr writer; rich would mangle "[agent-container]" as markup and wrap lines.
    print(msg, file=sys.stderr)


class Fatal(Exception):
    """Fatal error; CLI exits 1, wizard prints and returns to the menu."""


def die(msg: str) -> NoReturn:  # cli() turns Fatal into exit 1
    raise Fatal(msg)


# --- pure helpers ------------------------------------------------------------


def validate_name(name: str) -> str:
    """Validate the container short-name charset: ^[a-z0-9][a-z0-9_-]*$.

    >>> validate_name("acme")
    'acme'
    >>> validate_name("my-box")
    'my-box'
    >>> validate_name("Bad")
    Traceback (most recent call last):
    ...
    Fatal: invalid <name> 'Bad'; must match [a-z0-9][a-z0-9_-]*
    """
    if not name:
        die("missing required <name> argument")
    if not NAME_RE.fullmatch(name):
        die(f"invalid <name> '{name}'; must match [a-z0-9][a-z0-9_-]*")
    return name


def container_name(name: str) -> str:
    """
    >>> container_name("acme")
    'agent-container-acme'
    """
    return f"{CONTAINER_PREFIX}{name}"


def volume_name(name: str) -> str:
    """
    >>> volume_name("acme")
    'agent-container-acme-workspace'
    """
    return f"{CONTAINER_PREFIX}{name}-workspace"


def claude_volume_name(name: str) -> str:
    """
    >>> claude_volume_name("acme")
    'agent-container-acme-claude'
    """
    return f"{CONTAINER_PREFIX}{name}-claude"


def codex_volume_name(name: str) -> str:
    """
    >>> codex_volume_name("acme")
    'agent-container-acme-codex'
    """
    return f"{CONTAINER_PREFIX}{name}-codex"


def pi_volume_name(name: str) -> str:
    """
    >>> pi_volume_name("acme")
    'agent-container-acme-pi'
    """
    return f"{CONTAINER_PREFIX}{name}-pi"


def shellenv_volume_name(name: str) -> str:
    """
    >>> shellenv_volume_name("acme")
    'agent-container-acme-shellenv'
    """
    return f"{CONTAINER_PREFIX}{name}-shellenv"


def tmux_volume_name(name: str) -> str:
    """Persists ~/.config/tmux (tmux.conf + tpm plugins) across down/up.

    >>> tmux_volume_name("acme")
    'agent-container-acme-tmux'
    """
    return f"{CONTAINER_PREFIX}{name}-tmux"


def ssh_volume_name(name: str) -> str:
    """Persists ~/.ssh across down/up: authorized_keys, known_hosts, and the
    dev-owned SSH host key under hostkeys/. Because the container is rootless
    (sshd runs as dev), the host key lives here rather than in root-owned
    /etc/ssh, so a container keeps a STABLE SSH identity across recreation.

    >>> ssh_volume_name("acme")
    'agent-container-acme-ssh'
    """
    return f"{CONTAINER_PREFIX}{name}-ssh"


def all_volume_mounts(name: str) -> list[str]:
    """The seven per-container '-v' volume args, in the canonical fixed order
    (workspace, claude, codex, pi, shellenv, tmux, ssh). pi-coding-agent's
    config/auth dir is ~/.pi (verified from
    the package: piConfig.configDir='.pi', getAgentDir()->~/.pi/agent).

    >>> all_volume_mounts("acme")
    ['agent-container-acme-workspace:/workspace', 'agent-container-acme-claude:/home/dev/.claude', 'agent-container-acme-codex:/home/dev/.codex', 'agent-container-acme-pi:/home/dev/.pi', 'agent-container-acme-shellenv:/home/dev/.agent-container', 'agent-container-acme-tmux:/home/dev/.config/tmux', 'agent-container-acme-ssh:/home/dev/.ssh']
    """
    return [
        f"{volume_name(name)}:/workspace",
        f"{claude_volume_name(name)}:/home/dev/.claude",
        f"{codex_volume_name(name)}:/home/dev/.codex",
        f"{pi_volume_name(name)}:/home/dev/.pi",
        f"{shellenv_volume_name(name)}:/home/dev/.agent-container",
        f"{tmux_volume_name(name)}:/home/dev/.config/tmux",
        f"{ssh_volume_name(name)}:/home/dev/.ssh",
    ]


def per_container_volumes(name: str) -> list[str]:
    """All per-container volume NAMES, canonical order; used by --purge.

    >>> per_container_volumes("acme")
    ['agent-container-acme-workspace', 'agent-container-acme-claude', 'agent-container-acme-codex', 'agent-container-acme-pi', 'agent-container-acme-shellenv', 'agent-container-acme-tmux', 'agent-container-acme-ssh']
    """
    return [
        volume_name(name),
        claude_volume_name(name),
        codex_volume_name(name),
        pi_volume_name(name),
        shellenv_volume_name(name),
        tmux_volume_name(name),
        ssh_volume_name(name),
    ]


def resolve_bind_mount(spec: str) -> str:
    """Resolve a '--mount HOSTDIR[:CONTAINERPATH]' spec into an absolute,
    read-write '-v' value.

    HOSTDIR is resolved to an absolute real path (Path.resolve()) and must be
    an existing directory. CONTAINERPATH defaults to /workspace/<basename>; if
    given explicitly it must be absolute. No secret/env value ever reaches argv.

    Lima prerequisite: the host dir must sit under a path the Lima VM exposes
    WRITABLE (set `writable: true` on the relevant mount and restart Lima),
    otherwise the bind is read-only inside the VM.
    """
    host, sep, container = spec.partition(":")
    if not host:
        die(f"--mount: empty host directory in '{spec}'")
    p = Path(host)
    if not p.is_dir():
        die(f"--mount: host path '{host}' does not exist or is not a directory")
    abs_host = str(p.resolve())
    if sep:  # an explicit ':' was present
        if not container:
            die(f"--mount: empty container path in '{spec}'")
        if not container.startswith("/"):
            die(f"--mount: container path '{container}' must be absolute")
    else:
        container = f"/workspace/{Path(abs_host).name}"
    return f"{abs_host}:{container}"


# Container-side paths the entrypoint reads injected SSH material from (bind
# mounts). Kept in sync with INJECT_DIR in entrypoint.sh.
INJECT_HOST_KEY_PATH = "/run/agent-container/ssh_host_ed25519_key"
INJECT_AUTHORIZED_KEYS_PATH = "/run/agent-container/authorized_keys"


def validate_private_key(path: Path) -> None:
    """Fail fast if `path` is not a readable, unencrypted OpenSSH private key.
    stdin is closed so an encrypted key fails cleanly instead of prompting."""
    r = subprocess.run(
        ["ssh-keygen", "-y", "-f", str(path)],
        capture_output=True,
        text=True,
        stdin=subprocess.DEVNULL,
    )
    if r.returncode != 0:
        die(
            f"--host-key: {path} is not a valid, unencrypted OpenSSH private key "
            f"({r.stderr.strip() or 'ssh-keygen validation failed'})"
        )


def resolve_ssh_injection(
    name: str, host_key: Path | None, authorized_keys: list[Path]
) -> list[str]:
    """Resolve `up --host-key/--authorized-key` into read-only '-v' bind specs.

    The entrypoint installs whatever appears at the container-side inject paths
    onto the persisted ~/.ssh volume at boot. Both the host key and the
    concatenated authorized_keys are staged as per-container files under
    STATE_DIR and bind-mounted from there, so the bind survives a container
    restart (a temp file would vanish and leave an empty mount).

    The staged files are mode 0644 on purpose: the entrypoint reads them as the
    unprivileged `dev` user, whose uid need NOT equal the host uid that ran `up`
    (it does not on CI runners / any host with a different login uid). Docker has
    no bind-mount chown (unlike podman's `:U`), so an 0600 file owned by the host
    user is unreadable inside the container — the read fails and, under the
    entrypoint's `set -e`, crash-loops sshd. 0644 is safe: STATE_DIR is locked to
    0700 so no other HOST user can traverse to these files, while the container
    reads each file inode directly through the mount (host dir perms don't gate
    it). authorized_keys are public regardless; the host key stays dir-protected.
    """
    specs: list[str] = []
    STATE_DIR.mkdir(parents=True, exist_ok=True)
    STATE_DIR.chmod(0o700)
    if host_key is not None:
        hk = host_key.expanduser()
        if not hk.is_file():
            die(f"--host-key: {host_key} does not exist or is not a file")
        validate_private_key(hk)
        hk_file = STATE_DIR / f"{name}.host_key"
        hk_file.write_bytes(hk.read_bytes())
        hk_file.chmod(0o644)
        specs.append(f"{hk_file}:{INJECT_HOST_KEY_PATH}:ro")
    if authorized_keys:
        blob = ""
        for p in authorized_keys:
            pk = p.expanduser()
            if not pk.is_file():
                die(f"--authorized-key: {p} does not exist or is not a file")
            text = pk.read_text()
            blob += text if text.endswith("\n") else text + "\n"
        ak_file = STATE_DIR / f"{name}.authorized_keys"
        ak_file.write_text(blob)
        ak_file.chmod(0o644)
        specs.append(f"{ak_file}:{INJECT_AUTHORIZED_KEYS_PATH}:ro")
    return specs


def port_for_name(name: str) -> int:
    """Deterministic port: PORT_BASE + (sum of char codes mod PORT_RANGE).

    Name charset is ASCII-only (enforced by NAME_RE), so ord() matches
    bash's printf '%d' "'c" exactly.

    >>> port_for_name("acme")
    2206
    >>> port_for_name("my-box")
    2204
    """
    return PORT_BASE + (sum(ord(c) for c in name) % PORT_RANGE)


def name_to_key(name: str) -> str:
    """hosts.conf key prefix: hyphens -> underscores, uppercased.

    >>> name_to_key("my-box")
    'MY_BOX'
    >>> name_to_key("acme")
    'ACME'
    """
    return name.replace("-", "_").upper()


def resolve_ssh_user(override: str | None = None) -> str:
    """SSH user from --user / AGENT_CONTAINER_USER / 'dev', charset-checked so it can
    never be parsed by ssh as an option.

    >>> resolve_ssh_user("dev")
    'dev'
    >>> resolve_ssh_user("-oProxyCommand=x")
    Traceback (most recent call last):
    ...
    Fatal: invalid ssh user '-oProxyCommand=x'
    """
    user = override or os.environ.get("AGENT_CONTAINER_USER") or "dev"
    if not SSH_USER_RE.fullmatch(user):
        die(f"invalid ssh user '{user}'")
    return user


def validate_window(window: str) -> str:
    """tmux window name guard: it is embedded in the ssh remote shell command,
    so reject anything outside the safe charset before building it.

    >>> validate_window("agents")
    'agents'
    >>> validate_window("a; rm -rf ~")
    Traceback (most recent call last):
    ...
    Fatal: invalid tmux window 'a; rm -rf ~'; must match [A-Za-z0-9._-]+
    """
    if not WINDOW_RE.fullmatch(window):
        die(f"invalid tmux window '{window}'; must match [A-Za-z0-9._-]+")
    return window


def parse_kv_config(text: str) -> dict[str, str]:
    """Literal KEY=VALUE parser for hosts.conf. Never executes the file.

    Skips blank lines and # comments, strips an optional leading 'export ',
    splits on the first '=', honours one pair of matching quotes, and — like
    bash `source` — drops unquoted trailing ' # comment's (a '#' only starts
    a comment when preceded by whitespace; 'A=#lit' keeps the '#').

    >>> parse_kv_config("# c\\n\\nA=1\\nexport B='two'\\nC=\\"three\\"\\nD=a=b\\n")
    {'A': '1', 'B': 'two', 'C': 'three', 'D': 'a=b'}
    >>> parse_kv_config('H=vps.example.com # primary box\\nQ="a # b" # c\\nR=#lit\\n')
    {'H': 'vps.example.com', 'Q': 'a # b', 'R': '#lit'}
    """
    out: dict[str, str] = {}
    for line in text.splitlines():
        line = line.strip()
        if not line or line.startswith("#") or "=" not in line:
            continue
        if line.startswith("export "):
            line = line[len("export ") :].lstrip()
        key, _, value = line.partition("=")
        key = key.strip()
        value = value.strip()
        if value[:1] in ("'", '"'):
            # Quoted: take up to the matching close quote; anything after it
            # (e.g. a trailing comment) is ignored, as bash would.
            closing = value.find(value[0], 1)
            if closing != -1:
                value = value[1:closing]
        else:
            m = re.search(r"\s#", value)
            if m:
                value = value[: m.start()].rstrip()
        if key:
            out[key] = value
    return out


def env_file_candidates(name: str, cwd: Path) -> list[Path]:
    """Env-file resolution order (./.env, then per-name, then shared default).

    >>> [str(p) for p in env_file_candidates("acme", Path("/w"))][0]
    '/w/.env'
    """
    return [cwd / ".env", CONFIG_DIR / f"{name}.env", CONFIG_DIR / ".env"]


def resolve_env_file(name: str) -> Path | None:
    # Secret hygiene: only existence checks; contents are never read.
    for candidate in env_file_candidates(name, Path.cwd()):
        if candidate.is_file():
            return candidate
    return None


def state_file(name: str) -> Path:
    return STATE_DIR / f"{name}.port"


# --- per-host state (Feature 001) --------------------------------------------
# Runtime state is namespaced per host: STATE_DIR/<host>/<name>.{port,compose.yaml,
# host_key,authorized_keys}. Identity VALUES (container_name/port_for_name/volume
# names) are unchanged — only the state file LOCATION gains a host segment, so the
# same name may run on different hosts without collision (per-host daemons make
# project/volume collisions structurally impossible). The pre-per-host flat
# STATE_DIR/<name>.* files belong to the implicit 'local' host; migrate_flat_state()
# relocates them once, keeping the identity contract stable (Constitution IV).


def host_state_dir(host: str) -> Path:
    return STATE_DIR / host


def state_file_for(host: str, name: str) -> Path:
    return host_state_dir(host) / f"{name}.port"


def compose_file_path(host: str, name: str) -> Path:
    return host_state_dir(host) / f"{name}.compose.yaml"


@contextlib.contextmanager
def deployment_lock(host: str, name: str):
    """Serialize mutating lifecycle ops on one deployment (FR-017): a non-blocking
    advisory flock on <state>/<host>/<name>.lock. A second concurrent op on the
    same (host,name) fails fast rather than interleaving; independent deployments
    never contend. Read-only ops (list/logs) do NOT take this lock."""
    lock_dir = host_state_dir(host)
    lock_dir.mkdir(parents=True, exist_ok=True)
    f = (lock_dir / f"{name}.lock").open("w")
    try:
        try:
            fcntl.flock(f.fileno(), fcntl.LOCK_EX | fcntl.LOCK_NB)
        except OSError:
            die(f"another lifecycle operation is in progress for {name} on {host}")
        yield
    finally:
        with contextlib.suppress(OSError):
            fcntl.flock(f.fileno(), fcntl.LOCK_UN)
        f.close()


# Flat state artifacts written before per-host namespacing existed. Migrated into
# STATE_DIR/local/ on first host-aware access.
_FLAT_STATE_SUFFIXES = (".port", ".host_key", ".authorized_keys")


def migrate_flat_state() -> None:
    """One-time: relocate flat STATE_DIR/<name>.{port,host_key,authorized_keys}
    into STATE_DIR/local/ (the implicit host). Idempotent: skips when the target
    already exists and never overwrites. Safe to call repeatedly."""
    if not STATE_DIR.is_dir():
        return
    local_dir = host_state_dir(DEFAULT_HOST)
    for f in sorted(STATE_DIR.iterdir()):
        if not f.is_file() or f.suffix not in _FLAT_STATE_SUFFIXES:
            continue
        # Skip .compose.yaml (never existed flat) and anything already migrated.
        target = local_dir / f.name
        if target.exists():
            continue
        local_dir.mkdir(parents=True, exist_ok=True)
        os.replace(f, target)


# --- runtime -----------------------------------------------------------------

_hosts_warned = False

# Captured at startup so an abnormal ssh exit can't leave the terminal raw.
try:
    _ORIG_TERMIOS = termios.tcgetattr(sys.stdin.fileno()) if sys.stdin.isatty() else None
except Exception:
    _ORIG_TERMIOS = None


def detect_runtime() -> str:
    """Resolve the container runtime.

    AGENT_CONTAINER_RUNTIME (validated docker|podman, must be on PATH) always wins.
    Otherwise the default is platform-aware: on macOS the operator runs Lima +
    docker-cli, so prefer docker then podman; on Linux (the VPS) prefer podman
    then docker. Dies if neither is installed.
    """
    forced = os.environ.get("AGENT_CONTAINER_RUNTIME")
    if forced:
        if forced not in ("podman", "docker"):
            die(f"AGENT_CONTAINER_RUNTIME must be 'docker' or 'podman', got: {forced}")
        if not shutil.which(forced):
            die(f"AGENT_CONTAINER_RUNTIME={forced} but '{forced}' is not on PATH")
        return forced
    order = ("docker", "podman") if sys.platform.startswith("darwin") else ("podman", "docker")
    for rt in order:
        if shutil.which(rt):
            return rt
    die("neither 'podman' nor 'docker' on PATH (install one, or set AGENT_CONTAINER_RUNTIME)")


def query(argv: list[str], timeout: float | None = None) -> subprocess.CompletedProcess:
    return subprocess.run(argv, capture_output=True, text=True, timeout=timeout)


def container_running(rt: str, cname: str) -> bool:
    r = query([rt, "ps", "--filter", f"name=^{cname}$", "--format", "{{.Names}}"])
    return cname in r.stdout.splitlines()  # exact match, mirrors grep -qx


def container_exists(rt: str, cname: str) -> bool:
    r = query([rt, "ps", "-a", "--filter", f"name=^{cname}$", "--format", "{{.Names}}"])
    return cname in r.stdout.splitlines()


def ps_agent_container(
    rt: str, include_stopped: bool = False, strict: bool = False
) -> list[tuple[str, str, str, str]]:
    """(cname, image, status, uptime) rows for agent-container-* containers.

    Go-template + tab split, not --format json: podman and docker JSON
    shapes are incompatible, the template output is not. With strict=True (the
    `list` reconcile path) the call is bounded by a timeout and DIES on a nonzero
    exit — a failed local `ps` must not read as 'zero containers' (fail-closed,
    001-US3 lesson). Default (best-effort) behavior is unchanged for other callers.
    """
    argv = [rt, "ps"] + (["-a"] if include_stopped else [])
    argv += ["--format", "{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.RunningFor}}"]
    r = query(argv, timeout=15 if strict else None)
    if strict and r.returncode != 0:
        die(f"could not list local containers: {r.stderr.strip() or f'exit {r.returncode}'}")
    rows = []
    for line in r.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) == 4 and parts[0].startswith(CONTAINER_PREFIX):
            rows.append((parts[0], parts[1], parts[2], parts[3]))
    return rows


def image_exists(rt: str, tag: str) -> bool:
    # `image inspect` works on both podman and docker (podman's `image exists` does not).
    return query([rt, "image", "inspect", tag]).returncode == 0


def port_free(port: int) -> bool:
    # Wildcard bind: `-p {port}:2222` publishes on ALL interfaces (parity with
    # agent-container), so probe what will actually be bound, not just loopback.
    try:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.bind(("", port))
        return True
    except OSError:
        return False


def wait_port_released(port: int, timeout: float = 15.0) -> None:
    """Block until `port` is bindable again (or timeout elapses). `<rt> rm`
    returns before the daemon has finished tearing down the published-port
    forwarding, so an immediate `up` on the same name hits a false 'port in
    use'. Making `down` wait keeps a down+up (recreate) cycle on one name
    reliable. Best-effort: on timeout we simply return and let `up` report the
    real state."""
    deadline = time.monotonic() + timeout
    while time.monotonic() < deadline:
        if port_free(port):
            return
        time.sleep(0.25)


def write_state(host: str, name: str, port: int) -> None:
    host_state_dir(host).mkdir(parents=True, exist_ok=True)
    state_file_for(host, name).write_text(f"{port}\n")


def read_state_port(host: str, name: str) -> str | None:
    f = state_file_for(host, name)
    try:
        if not f.is_file():
            return None
        return f.read_text().strip() or None
    except OSError:  # unreadable / TOCTOU delete — degrade to unknown, never abort a listing
        return None


def clear_state(host: str, name: str) -> None:
    state_file_for(host, name).unlink(missing_ok=True)


def load_hosts_conf() -> dict[str, str]:
    global _hosts_warned
    if not HOSTS_CONF.is_file():
        return {}
    conf = parse_kv_config(HOSTS_CONF.read_text())
    if not _hosts_warned and any("$" in v or "`" in v for v in conf.values()):
        eprint(
            "[agent-container] WARNING: hosts.conf contains '$' or '`' values; "
            "agent-container does not perform shell expansion (agent-container does)."
        )
        _hosts_warned = True
    return conf


def hosts_entry(name: str) -> tuple[str, str] | None:
    """(host, port) from hosts.conf, or None unless BOTH keys are present."""
    conf = load_hosts_conf()
    key = name_to_key(name)
    host, port = conf.get(f"{key}_HOST"), conf.get(f"{key}_PORT")
    if host and port:
        return host, port
    return None


# --- host registry (Feature 001) ---------------------------------------------
# The registry (hosts.json) is the single source of truth for WHERE containers
# run, superseding the flat hosts.conf address book. Shape:
#   {"version": 1, "default": "<name>|null", "hosts": {"<name>": {Host}, ...}}
# A Host record: {driver, context, address, port?, provisioning, created_by_tool}.
# When hosts.json is absent, a legacy hosts.conf is read into read-only
# 'existing-ssh' (attach-only) hosts for a deprecation window. The registry is
# written atomically and its content is NEVER executed (parity with hosts.conf).


def _synthesize_legacy_registry() -> dict:
    """Read a legacy hosts.conf into attach-only 'existing-ssh' Host records.
    Not written back; removed after the deprecation window."""
    conf = load_hosts_conf()
    hosts: dict = {}
    for key in conf:
        if not key.endswith("_HOST"):
            continue
        base = key[: -len("_HOST")]
        host, port = conf[key], conf.get(f"{base}_PORT")
        if not (host and port):
            continue
        hosts[base.lower()] = {
            "driver": "existing-ssh",
            "context": "",
            "address": host,
            "port": port,
            "provisioning": None,
            "created_by_tool": False,
        }
    return {"version": REGISTRY_VERSION, "default": None, "hosts": hosts}


def load_registry() -> dict:
    """The host registry. Reads hosts.json when present, else synthesizes a
    read-only legacy registry from hosts.conf. Dies on a malformed file rather
    than silently losing host records."""
    if HOSTS_JSON.is_file():
        try:
            data = json.loads(HOSTS_JSON.read_text())
        except (json.JSONDecodeError, OSError) as e:
            die(f"invalid host registry at {HOSTS_JSON}: {e}")
        if not isinstance(data, dict) or not isinstance(data.get("hosts"), dict):
            die(f"invalid host registry at {HOSTS_JSON}: missing 'hosts' object")
        data.setdefault("version", REGISTRY_VERSION)
        data.setdefault("default", None)
        return data
    return _synthesize_legacy_registry()


def save_registry(reg: dict) -> None:
    """Persist the registry atomically (temp + os.replace) so a crash mid-write
    never truncates hosts.json."""
    CONFIG_DIR.mkdir(parents=True, exist_ok=True)
    tmp = HOSTS_JSON.with_name(HOSTS_JSON.name + ".tmp")
    tmp.write_text(json.dumps(reg, indent=2) + "\n")
    os.replace(tmp, HOSTS_JSON)


def registry_hosts(reg: dict) -> dict:
    hosts = reg.get("hosts")
    return hosts if isinstance(hosts, dict) else {}


def get_host(reg: dict, name: str) -> dict | None:
    return registry_hosts(reg).get(name)


def default_host_name(reg: dict) -> str | None:
    """The registry's default deploy target, or None."""
    d = reg.get("default")
    return d if isinstance(d, str) else None


def address_from_context(context: str) -> str:
    """Attach address implied by a runtime context: the host of an ssh:// context,
    else localhost (a local context publishes to the local machine).

    >>> address_from_context("ssh://root@203.0.113.7")
    '203.0.113.7'
    >>> address_from_context("ssh://ops@vps.example.com:22")
    'vps.example.com'
    >>> address_from_context("lima-docker")
    'localhost'
    """
    m = re.match(r"ssh://(?:[^@/]+@)?([^:/]+)", context)
    return m.group(1) if m else "localhost"


def probe_host_runtime(host: dict) -> str | None:
    """Best-effort capability probe: returns an error string if the host's
    runtime is not usable (binary missing / context unreachable), else None.
    Non-blocking by design — a host may be registered before its context is
    reachable (e.g. just-provisioned); the deploy path fails hard later if still
    unusable. Bounded so a dead ssh:// context can't hang registration."""
    ensure_tunnel(host)
    try:
        argv = driver_runtime_argv(host) + ["version"]
    except Fatal:
        return "driver is attach-only"
    if not shutil.which(argv[0]):
        return f"'{argv[0]}' is not on PATH"
    try:
        r = subprocess.run(argv, capture_output=True, text=True, timeout=8)
    except (subprocess.TimeoutExpired, OSError) as e:
        return f"runtime probe failed: {e}"
    if r.returncode != 0:
        return (r.stderr.strip() or "runtime probe failed").splitlines()[0][:160]
    return None


def cli_host_add(
    name: str,
    driver: str,
    context: str | None,
    address: str | None,
    make_default: bool,
    *,
    provider: str | None = None,
    create: bool = False,
    reuse: bool = False,
    server_type: str | None = None,
    location: str | None = None,
    ssh_key: str | None = None,
    ssh_pubkey: Path | None = None,
) -> None:
    """Register a container-runtime host. Without --provider, a local/remote
    docker/podman context. With --provider, either allocate a cloud server
    (--create, billable) or register an existing one (--reuse)."""
    validate_name(name)
    if provider:
        if create == reuse:
            die("with --provider choose exactly one of --create (allocate) or --reuse (existing)")
        if create:
            record = provision_host(
                provider,
                name,
                server_type=server_type,
                location=location,
                ssh_key=ssh_key,
                ssh_pubkey=ssh_pubkey,
            )
        else:  # --reuse: register an operator-supplied server; never destroyable
            if not context:
                die("--reuse needs --docker-context (ssh://user@host or an existing context name)")
            if context.startswith("ssh://"):
                # A raw ssh:// URL is not a valid `docker --context` value; wrap it
                # in a named local context so the deploy path works unchanged.
                ctx_name = docker_context_name(name)
                docker_context_create(ctx_name, context)
                reuse_ctx, reuse_addr = ctx_name, address or address_from_context(context)
            else:
                reuse_ctx, reuse_addr = context, address or "localhost"
            record = {
                "driver": "docker",
                "context": reuse_ctx,
                "address": reuse_addr,
                "provisioning": {"provider": provider, "created": False},
                "created_by_tool": False,
            }
    else:
        if driver not in ("docker", "podman"):
            die(f"--driver must be 'docker' or 'podman' (got '{driver}')")
        if not context:
            opt = "--docker-context" if driver == "docker" else "--connection"
            die(f"host add: a {driver} host needs {opt} <context>")
        record = {
            "driver": driver,
            "context": context,
            "address": address or address_from_context(context),
            "provisioning": None,
            "created_by_tool": False,
        }
    reg = load_registry()
    hosts = registry_hosts(reg)
    existed = name in hosts
    hosts[name] = record
    reg["hosts"] = hosts
    if make_default or default_host_name(reg) is None:
        reg["default"] = name
    save_registry(reg)
    verb = "updated" if existed else "registered"
    is_default = " [default]" if default_host_name(reg) == name else ""
    addr = f", address={record['address']}" if record.get("address") else ""
    log(
        f"{verb} host '{name}' (driver={record['driver']}, "
        f"context={record['context']}{addr}){is_default}"
    )
    err = probe_host_runtime(record)
    if err is not None:
        warn(f"host '{name}' registered but its runtime is not usable yet: {err}")


def do_host_ls(as_json: bool) -> None:
    reg = load_registry()
    hosts = registry_hosts(reg)
    default = default_host_name(reg)
    if as_json:
        print(json.dumps({"default": default, "hosts": hosts}, indent=2))
        return
    if not hosts:
        log("no hosts registered (add one: agent-container host add <name> --docker-context <ctx>)")
        return
    table = Table(show_header=True, header_style="bold", box=None, pad_edge=False)
    for col in ("NAME", "DRIVER", "CONTEXT", "ADDRESS", "DEFAULT"):
        table.add_column(col)
    for hname in sorted(hosts):
        h = hosts[hname]
        table.add_row(
            hname,
            str(h.get("driver", "?")),
            str(h.get("context") or "-"),
            str(h.get("address") or "-"),
            "*" if hname == default else "",
        )
    console.print(table)


def do_host_show(name: str, as_json: bool) -> None:
    """Show one host's full record. Read-only: never saves, never fetches a token,
    never contacts a daemon (the record holds no secret — only ids)."""
    validate_name(name)
    reg = load_registry()
    h = get_host(reg, name)
    if h is None:
        die(f"no host named '{name}' (see: agent-container host ls)")
    is_default = default_host_name(reg) == name
    if as_json:
        print(json.dumps({"name": name, "default": is_default, **h}, indent=2))
        return
    prov = h.get("provisioning")
    table = Table(show_header=False, box=None, pad_edge=False)
    table.add_column(style="bold")
    table.add_column()
    table.add_row("name", name)
    table.add_row("driver", str(h.get("driver", "?")))
    table.add_row("context", str(h.get("context") or "-"))
    table.add_row("address", str(h.get("address") or "-"))
    table.add_row("default", "yes" if is_default else "no")
    table.add_row("created_by_tool", "yes" if h.get("created_by_tool") else "no")
    table.add_row("provisioning", json.dumps(prov) if prov else "-")
    console.print(table)


def cli_host_rm(name: str, destroy: bool, yes: bool) -> None:
    """Remove a host from the registry. Without --destroy, only the registration
    is removed — infrastructure is NEVER touched (FR-010). With --destroy, also
    deprovision the cloud server, but ONLY when: the tool created it, its provider
    has a deprovisioner, and no container remains on it (FR-008/009/010, SC-005).
    The registry entry is removed only AFTER a successful deprovision, so a partial
    teardown leaves the record for an idempotent retry."""
    validate_name(name)
    reg = load_registry()
    h = get_host(reg, name)
    if h is None:
        die(f"no host named '{name}' (see: agent-container host ls)")
    if destroy:
        if not h.get("created_by_tool"):
            die(
                f"refusing --destroy on '{name}': this tool did not create its server "
                f"(FR-010). Remove just the registry entry with: agent-container host rm {name}"
            )
        prov = h.get("provisioning") or {}
        if prov.get("provider") != "hetzner":
            die(
                f"cannot --destroy '{name}': no deprovisioner for provider '{prov.get('provider')}'"
            )
        # FR-009 / SC-005: never destroy a server that still hosts containers.
        # Fail-CLOSED (assert_host_empty): only a container listing that provably
        # succeeded and is empty may proceed — a down tunnel or an unreachable
        # daemon refuses rather than being read as "empty".
        assert_host_empty(h)
        if not yes:
            if not is_tty():
                eprint(
                    f"[agent-container] refusing to destroy host '{name}' "
                    f"without -y/--yes on a non-TTY"
                )
                raise typer.Exit(2)
            if not questionary.confirm(
                f"destroy the cloud server for host '{name}' (irreversible)?", default=False
            ).ask():
                log("aborted")
                return
        token = _hcloud_token()  # only here, only after every refusal + the confirm
        provisioner_destroy(h, token)
    elif h.get("created_by_tool"):
        warn(
            f"host '{name}' was provisioned by this tool; its cloud server is left "
            f"running (and billable). Use 'host rm --destroy {name}' to deallocate it."
        )
    was_default = default_host_name(reg) == name
    hosts = registry_hosts(reg)
    hosts.pop(name, None)
    reg["hosts"] = hosts
    if was_default:  # never leave the default pointing at a removed host
        reg["default"] = next(iter(sorted(hosts)), None)
    save_registry(reg)
    log(f"removed host '{name}'" + (" and destroyed its server" if destroy else ""))


# --- Hetzner provisioner (Feature 001, US2) ----------------------------------
# A provisioner allocates a cloud server and yields a plain docker-driver host,
# so all provider specifics stay confined here (adding a provider is a new
# provisioner, not a change to build/run/attach). Stdlib urllib only — no hcloud
# SDK (Constitution VI). The API token is read from HCLOUD_TOKEN at call time and
# used ONLY in the Bearer header — never on argv, never baked, never persisted to
# hosts.json, never logged (Constitution III). Server allocation is billable and
# happens ONLY on an explicit `--create`.
#
# NOT LIVE-VALIDATED against real Hetzner in unit tests; the create/response
# shapes, cloud-init timing, and ssh-context behavior are exercised by the opt-in
# tokened acceptance test (bin/tests, -m acceptance) an operator runs manually.

HCLOUD_API = "https://api.hetzner.cloud/v1"
HETZNER_DEFAULT_SERVER_TYPE = "cax11"  # cheapest ARM shared vCPU; override with --server-type
HETZNER_DEFAULT_LOCATION = "nbg1"
HETZNER_IMAGE = "debian-12"
HETZNER_CODENAME = "bookworm"  # debian-12 suite on download.docker.com (pinned to HETZNER_IMAGE)
# Hetzner server names must be RFC-1123 labels (no underscore; <=63) — stricter than validate_name.
HETZNER_NAME_RE = re.compile(r"[a-z0-9]([a-z0-9-]{0,61}[a-z0-9])?$")


class _NoAuthRedirect(urllib.request.HTTPRedirectHandler):
    """Strip the Bearer header on any redirect so the token never follows a
    cross-host Location (defense in depth; Hetzner does not redirect)."""

    def redirect_request(self, req, fp, code, msg, headers, newurl):
        new = super().redirect_request(req, fp, code, msg, headers, newurl)
        if new is not None:
            new.remove_header("Authorization")
        return new


_HCLOUD_OPENER = urllib.request.build_opener(_NoAuthRedirect())
_HCLOUD_IDEMPOTENT = {"GET", "DELETE"}  # safe to retry; POST is not (double-create risk)


def _hcloud_token() -> str:
    """The Hetzner API token from HCLOUD_TOKEN. Dies (before any HTTP call) if
    unset. Never returned into a Host record, argv, or log."""
    tok = os.environ.get("HCLOUD_TOKEN")
    if not tok:
        die(
            "HCLOUD_TOKEN is not set — export it (e.g. from your secret store) before "
            "provisioning. It is read from the environment only, never stored or on argv."
        )
    return tok


def _retry_delay(exc: urllib.error.HTTPError | None, attempt: int) -> float:
    """Backoff: honor Retry-After on a 429/503 if present, else exponential (cap 15s)."""
    if exc is not None and exc.headers:
        ra = exc.headers.get("Retry-After")
        if ra and ra.strip().isdigit():
            return min(float(ra), 30.0)
    return min(2.0**attempt, 15.0)


def _hcloud_request(
    method: str,
    path: str,
    token: str,
    body: dict | None = None,
    *,
    timeout: float = 15.0,
    retries: int = 3,
) -> tuple[int, dict]:
    """One Hetzner REST call via stdlib urllib. Bearer header (stripped on redirect),
    JSON body, explicit timeout. Transient 429/5xx and transport errors are retried
    with backoff for idempotent methods (GET/DELETE) — never POST, to avoid double
    allocation. Non-2xx becomes Fatal with the API's error.code/message; the token
    never appears in the message."""
    url = f"{HCLOUD_API}{path}"
    if not url.startswith("https://"):  # defense: scheme is always the fixed HCLOUD_API host
        die(f"refusing non-https Hetzner API URL: {url}")
    data = json.dumps(body).encode() if body is not None else None
    retryable = method in _HCLOUD_IDEMPOTENT
    attempt = 0
    while True:
        attempt += 1
        req = urllib.request.Request(url, data=data, method=method)
        req.add_header("Authorization", f"Bearer {token}")
        req.add_header("Content-Type", "application/json")
        try:
            # opener.open (not urlopen); url is the fixed https HCLOUD_API base.
            with _HCLOUD_OPENER.open(req, timeout=timeout) as resp:
                raw = resp.read()
                return resp.status, (json.loads(raw) if raw else {})
        except urllib.error.HTTPError as e:
            if (e.code == 429 or e.code >= 500) and retryable and attempt <= retries:
                time.sleep(_retry_delay(e, attempt))
                continue
            raw = e.read()
            try:
                err = json.loads(raw).get("error", {})
                msg = f"{err.get('code', e.code)}: {err.get('message', e.reason)}"
            except ValueError, AttributeError:
                msg = f"{e.code}: {e.reason}"
            die(f"Hetzner API {method} {path} failed ({msg})")
        except (urllib.error.URLError, TimeoutError, OSError) as e:
            if retryable and attempt <= retries:
                time.sleep(_retry_delay(None, attempt))
                continue
            die(f"Hetzner API {method} {path} unreachable ({e})")


def resolve_operator_pubkey(override: Path | None = None) -> str:
    """The operator SSH PUBLIC key to authorize on the new server (so the operator's
    docker ssh:// context + attach can reach it). --ssh-pubkey wins, else the first
    of ~/.ssh/id_ed25519.pub / id_rsa.pub."""

    def _public(text: str, src: str) -> str:
        # Guard against embedding a PRIVATE key in user_data (it would land in the
        # server's cloud-init logs). Public keys start with a known type prefix.
        if "PRIVATE KEY" in text or not re.match(r"(ssh-|ecdsa-|sk-)", text):
            die(f"{src} is not an SSH PUBLIC key (expected ssh-ed25519/ssh-rsa/…)")
        return text

    if override is not None:
        p = override.expanduser()
        if not p.is_file():
            die(f"--ssh-pubkey: {override} does not exist")
        return _public(p.read_text().strip(), f"--ssh-pubkey {override}")
    for cand in ("id_ed25519.pub", "id_rsa.pub"):
        p = Path.home() / ".ssh" / cand
        if p.is_file():
            return _public(p.read_text().strip(), str(p))
    die(
        "no SSH public key found (~/.ssh/id_ed25519.pub or id_rsa.pub); pass "
        "--ssh-pubkey <path> so the server authorizes a key you hold"
    )


def hetzner_build_user_data() -> str:
    """cloud-init that installs Docker + the Compose v2 plugin on a fresh Debian 12
    server. Passed verbatim as user_data (plain YAML, no base64)."""
    # DOCKER INSTALL ONLY — key authorization is handled via Hetzner's ssh_keys
    # API (injected into root at provision time), NOT cloud-init: on the Hetzner
    # debian-12 image cloud-init's top-level `ssh_authorized_keys` does NOT
    # authorize root (verified live — root rejects the key), so `root@ip` auth
    # would fail and the docker-over-ssh poll would never succeed. The apt suite
    # is pinned to HETZNER_CODENAME (not a nested $(...) whose quoting broke).
    return (
        "#cloud-config\n"
        "package_update: true\n"
        "packages:\n"
        "  - ca-certificates\n"
        "  - curl\n"
        "runcmd:\n"
        "  - install -m 0755 -d /etc/apt/keyrings\n"
        "  - curl -fsSL --retry 5 --retry-connrefused https://download.docker.com/linux/debian/gpg -o /etc/apt/keyrings/docker.asc\n"
        "  - chmod a+r /etc/apt/keyrings/docker.asc\n"
        f'  - echo "deb [arch=$(dpkg --print-architecture) signed-by=/etc/apt/keyrings/docker.asc] https://download.docker.com/linux/debian {HETZNER_CODENAME} stable" > /etc/apt/sources.list.d/docker.list\n'
        "  - apt-get update -o DPkg::Lock::Timeout=600\n"
        "  - DEBIAN_FRONTEND=noninteractive apt-get install -y -o DPkg::Lock::Timeout=600 docker-ce docker-ce-cli containerd.io docker-buildx-plugin docker-compose-plugin\n"
        "  - systemctl enable --now docker\n"
    )


def hetzner_create_server(
    name: str,
    server_type: str,
    location: str,
    user_data: str,
    token: str,
    ssh_keys: list,
) -> dict:
    """POST /servers (pure allocation; the caller owns cleanup of a half-provisioned
    server). ssh_keys (ids/names) are injected into root by Hetzner. Returns the
    parsed create response (server.id, server.public_net.ipv4.ip)."""
    body: dict = {
        "name": name,
        "server_type": server_type,
        "location": location,
        "image": HETZNER_IMAGE,
        "start_after_create": True,
        "public_net": {"enable_ipv4": True, "enable_ipv6": True},
        "user_data": user_data,
        "ssh_keys": ssh_keys,
    }
    _status, resp = _hcloud_request("POST", "/servers", token, body, timeout=30.0)
    return resp


def hetzner_ensure_ssh_key(name: str, pubkey: str, token: str) -> tuple[int, bool]:
    """Ensure `pubkey` exists as a Hetzner project SSH key and return (id, created_by_us).
    Reuses an existing key with the same public key (so we never delete a key the
    operator uses elsewhere); otherwise uploads it under `name` (replacing a stale
    same-named key first). Injecting via the API is what actually authorizes root —
    cloud-init's ssh_authorized_keys does not on this image."""
    want = pubkey.split()[:2]  # match on type + base64, ignore the trailing comment
    _status, existing = _hcloud_request("GET", "/ssh_keys", token)
    keys = existing.get("ssh_keys", [])
    for k in keys:
        if k.get("public_key", "").split()[:2] == want:
            return int(k["id"]), False
    for k in keys:  # drop a stale key with our name (e.g. a prior failed run)
        if k.get("name") == name:
            _hcloud_request("DELETE", f"/ssh_keys/{k['id']}", token)
    _status, resp = _hcloud_request(
        "POST", "/ssh_keys", token, {"name": name, "public_key": pubkey}
    )
    return int(resp["ssh_key"]["id"]), True


def hetzner_delete_ssh_key(key_id: int, token: str) -> None:
    """DELETE /ssh_keys/<id>, idempotent + best-effort (only for keys we uploaded)."""
    try:
        _hcloud_request("DELETE", f"/ssh_keys/{key_id}", token)
    except Fatal as e:
        if "not_found" not in str(e):
            warn(f"could not delete Hetzner ssh key {key_id}: {e}")


def hetzner_get_server(server_id: int, token: str) -> dict:
    _status, resp = _hcloud_request("GET", f"/servers/{server_id}", token)
    return resp.get("server", {})


def hetzner_find_server_by_name(name: str, token: str) -> dict | None:
    """Best-effort lookup by exact name — used to reconcile a server that a POST
    created but whose response was lost, so we can still delete it (no orphan)."""
    try:
        _status, resp = _hcloud_request("GET", f"/servers?name={name}", token)
    except Fatal:
        return None
    servers = resp.get("servers") or []
    return servers[0] if servers else None


def hetzner_delete_server(server_id: int, token: str, *, strict: bool = False) -> None:
    """DELETE /servers/<id>, idempotent (404 => already gone). Best-effort by
    default so a failed *rollback* (cleanup_on_failure) never masks the original
    error; but strict=True on the deliberate `host rm --destroy` teardown RAISES on
    a non-404 failure, so the caller keeps the host registered for a retry rather
    than silently orphaning a live billable server (SC-009)."""
    try:
        _hcloud_request("DELETE", f"/servers/{server_id}", token, timeout=30.0)
    except Fatal as e:
        if "not_found" in str(e):
            return
        if strict:
            raise
        warn(f"could not delete Hetzner server {server_id}: {e}")


def remove_automation_key(name: str) -> None:
    """Delete the local file-based automation keypair for host `name`."""
    key = automation_key_path(name)
    key.unlink(missing_ok=True)
    (key.parent / f"{key.name}.pub").unlink(missing_ok=True)


def cleanup_on_failure(
    server_id: int | None,
    token: str,
    *,
    ssh_key_ids: tuple[int | None, ...] = (),
    name: str | None = None,
) -> None:
    """Destroy a half-provisioned server (and any keys we uploaded, plus the local
    automation key + its tunnel) so a failed `host add` leaves nothing billable or
    dangling (FR-011). No-op for anything not allocated yet."""
    if server_id is not None:
        warn(f"cleaning up half-provisioned Hetzner server {server_id}")
        hetzner_delete_server(server_id, token)
    for key_id in ssh_key_ids:
        if isinstance(key_id, int):
            hetzner_delete_ssh_key(key_id, token)
    if name:
        _close_tunnel(name)
        remove_automation_key(name)


def seed_known_hosts(ip: str) -> None:
    """Pin the new server's host key so `docker context` over ssh:// and `attach`
    connect non-interactively. First drop any stale key for this IP — Hetzner
    recycles public IPv4s, and a stale entry makes ssh refuse the new host
    (REMOTE HOST IDENTIFICATION HAS CHANGED), which would hang the docker poll."""
    kh = Path.home() / ".ssh" / "known_hosts"
    kh.parent.mkdir(mode=0o700, exist_ok=True)
    query(["ssh-keygen", "-R", ip])  # remove any recycled/stale key + dedup
    scan = query(["ssh-keyscan", "-t", "ed25519", "-T", "10", ip])
    if scan.returncode == 0 and scan.stdout:
        with kh.open("a") as f:
            f.write(scan.stdout)


def docker_context_name(name: str) -> str:
    return f"{CONTAINER_PREFIX}{name}"  # agent-container-<name>, mirrors the project name


def docker_context_create(ctx: str, endpoint: str) -> None:
    """Create (idempotently) a named local docker context pointing at an ssh://
    endpoint, so `docker --context <ctx> compose` (which honors --context) builds/
    runs on the server. `endpoint` is a full docker host, e.g. ssh://root@1.2.3.4."""
    query(["docker", "context", "rm", "-f", ctx])  # tolerate a stale one
    r = query(["docker", "context", "create", ctx, "--docker", f"host={endpoint}"])
    if r.returncode != 0:
        die(f"could not create docker context {ctx}: {r.stderr.strip()}")


def _docker_ctx_ready(ctx: str) -> tuple[bool, str]:
    """`docker --context <ctx> version` exits 0, with a HARD timeout so a first-
    connect ssh host-key prompt can never hang the poll forever (BatchMode is not
    guaranteed on docker's ssh helper; known_hosts is seeded first as the primary
    guard, this is the backstop). Returns (ready, why) — `why` carries the last
    failure reason (ssh host-key/auth vs. daemon-not-up) so a poll timeout can say
    what actually kept failing instead of a blind 'never became reachable'."""
    try:
        r = subprocess.run(
            ["docker", "--context", ctx, "version"],
            capture_output=True,
            text=True,
            stdin=subprocess.DEVNULL,
            timeout=20,
        )
        if r.returncode == 0:
            return True, ""
        return False, (r.stderr.strip() or r.stdout.strip() or f"exit {r.returncode}")
    except subprocess.TimeoutExpired:
        return False, "docker version timed out after 20s"
    except OSError as e:
        return False, f"docker invocation failed: {e}"


# --- provisioned-host automation identity + ssh socket-forward ---------------
# A provisioned host authorizes a dedicated, FILE-based automation key alongside
# the operator key. All of the tool's docker traffic runs over an ssh LOCAL-socket
# forward that presents that key with every ssh option passed as a command-line
# arg (-i / -o …) — so it signs unattended regardless of the operator's ~/.ssh/
# config or agent (e.g. a 1Password key that needs interactive approval per use).
# The operator key stays authorized for interactive `attach`. The docker context
# for such a host points at unix://<sock> (not ssh://), so every existing
# `docker --context` deploy/ps/down path is unchanged — only the sock must be live.

_TUNNELS: dict[str, subprocess.Popen] = {}


def host_name_from_context(ctx: str) -> str:
    return ctx[len(CONTAINER_PREFIX) :] if ctx.startswith(CONTAINER_PREFIX) else ctx


def automation_key_path(name: str) -> Path:
    """File-based ed25519 automation private key for host `name` (0600)."""
    return host_state_dir(name) / "automation_key"


def _socket_base() -> Path:
    """Short, user-private base dir for the docker forward sockets. NOT under
    STATE_DIR: a unix socket's sun_path is capped (~104 bytes on macOS) and a deep
    XDG_STATE_HOME (e.g. a pytest tmp dir) would overflow it. Prefer XDG_RUNTIME_DIR
    (the standard socket home on Linux); else a per-uid dir under the system tmp."""
    rt = os.environ.get("XDG_RUNTIME_DIR")
    base = Path(rt) if rt else Path(tempfile.gettempdir()) / f"agent-container-{os.getuid()}"
    base.mkdir(mode=0o700, parents=True, exist_ok=True)
    return base


def daemon_socket_path(name: str) -> Path:
    """Local unix socket the docker context connects to (hashed to a short name)."""
    h = hashlib.sha256(name.encode()).hexdigest()[:12]
    return _socket_base() / f"ac-{h}.sock"


def generate_automation_key(name: str) -> str:
    """Create (idempotently) the automation keypair and return its public key."""
    key = automation_key_path(name)
    key.parent.mkdir(parents=True, exist_ok=True)
    if not key.exists():
        query(
            ["ssh-keygen", "-t", "ed25519", "-N", "", "-q",
             "-C", f"agent-container-{name}-automation", "-f", str(key)]
        )  # fmt: skip
        key.chmod(0o600)
    return (key.parent / f"{key.name}.pub").read_text().strip()


def _forward_argv(ip: str, key: Path, sock: Path) -> list[str]:
    """ssh LOCAL-socket forward to the remote docker daemon — every option a literal
    CLI arg (no ~/.ssh/config, no agent): the automation key only, agent disabled,
    host key auto-pinned. `-N` (no remote shell); run foreground so we own the pid."""
    kh = Path.home() / ".ssh" / "known_hosts"
    return [
        shutil.which("ssh") or "ssh",
        "-i", str(key),
        "-o", "IdentitiesOnly=yes",
        "-o", "IdentityAgent=none",
        "-o", "StrictHostKeyChecking=accept-new",
        "-o", f"UserKnownHostsFile={kh}",
        "-o", "ExitOnForwardFailure=yes",
        "-o", "ServerAliveInterval=15",
        "-o", "BatchMode=yes",
        "-N", "-L", f"{sock}:/var/run/docker.sock", f"root@{ip}",
    ]  # fmt: skip


def _close_tunnel(name: str) -> None:
    proc = _TUNNELS.pop(name, None)
    if proc and proc.poll() is None:
        proc.terminate()
        try:
            proc.wait(timeout=5)
        except subprocess.TimeoutExpired:
            proc.kill()
    daemon_socket_path(name).unlink(missing_ok=True)


atexit.register(lambda: [_close_tunnel(n) for n in list(_TUNNELS)])


def _start_tunnel(name: str, ip: str) -> subprocess.Popen:
    """(Re)start the docker socket-forward for host `name`; idempotent — a running
    tunnel is reused. A dead one (auth not yet injected during cloud-init) is
    replaced so the readiness poll self-heals."""
    proc = _TUNNELS.get(name)
    if proc and proc.poll() is None:
        return proc
    sock = daemon_socket_path(name)
    sock.parent.mkdir(parents=True, exist_ok=True)
    sock.unlink(missing_ok=True)
    proc = subprocess.Popen(
        _forward_argv(ip, automation_key_path(name), sock),
        stdin=subprocess.DEVNULL,
        stdout=subprocess.DEVNULL,
        stderr=subprocess.DEVNULL,
    )
    _TUNNELS[name] = proc
    return proc


def ensure_tunnel(host: dict, *, required: bool = False) -> None:
    """For a provisioned (ssh-forward) host, ensure its docker socket-forward is up
    for the life of this process, waiting briefly for the socket to bind. No-op for
    local / user-supplied-context / existing-ssh hosts. With required=True (the
    teardown safety path), DIE if the forward cannot be established within the
    deadline — a down tunnel must never be mistaken for a reachable-but-empty
    daemon (SC-005)."""
    prov = host.get("provisioning") or {}
    if prov.get("connection") != "ssh-forward":
        return
    name = host_name_from_context(host.get("context") or "")
    ip = host.get("address")
    if not name or not ip:
        return
    sock = daemon_socket_path(name)
    proc = _start_tunnel(name, ip)
    deadline = time.monotonic() + 15
    while time.monotonic() < deadline:
        if sock.exists():
            return
        if proc.poll() is not None:  # tunnel died — retry (idempotent)
            proc = _start_tunnel(name, ip)
        time.sleep(0.3)
    if required:
        die(
            f"could not establish the docker socket-forward to host '{name}' ({ip}); "
            f"refusing to proceed"
        )


def wait_until_reachable(
    ip: str,
    ctx: str,
    *,
    ssh_timeout: float = 150.0,
    docker_timeout: float = 420.0,
    poll_interval: float = 5.0,
) -> None:
    """Two independent budgets: SSH port open (cloud-init authorized the key), THEN
    — with its own fresh, larger budget — `docker version` succeeds over the socket-
    forward (docker install from cloud-init finishes, minutes on a small node). A
    shared deadline would starve the docker phase when boot is slow. Both budgets
    are env-tunable (AGENT_CONTAINER_HZ_SSH_TIMEOUT / _DOCKER_TIMEOUT). The forward
    is (re)established each poll: early on the automation key may not be injected
    yet, so the tunnel dies and is retried until auth succeeds."""
    ssh_timeout = float(os.environ.get("AGENT_CONTAINER_HZ_SSH_TIMEOUT", ssh_timeout))
    docker_timeout = float(os.environ.get("AGENT_CONTAINER_HZ_DOCKER_TIMEOUT", docker_timeout))
    name = host_name_from_context(ctx)
    ssh_deadline = time.monotonic() + ssh_timeout
    while time.monotonic() < ssh_deadline:
        with socket.socket(socket.AF_INET, socket.SOCK_STREAM) as s:
            s.settimeout(5)
            if s.connect_ex((ip, 22)) == 0:
                break
        time.sleep(poll_interval)
    else:
        die(f"server {ip}: SSH never became reachable within {ssh_timeout:.0f}s")
    seed_known_hosts(ip)
    docker_deadline = time.monotonic() + docker_timeout
    last_why = "(no probe completed)"
    while time.monotonic() < docker_deadline:
        _start_tunnel(name, ip)  # (re)establish; heals a tunnel that died pre-auth
        ready, last_why = _docker_ctx_ready(ctx)
        if ready:
            return
        time.sleep(poll_interval)
    die(
        f"server {ip}: docker never became reachable within {docker_timeout:.0f}s "
        f"(cloud-init?); last probe error: {last_why}"
    )


def provisioner_create(
    name: str,
    server_type: str,
    location: str,
    operator_pubkey: str,
    extra_ssh_key: str | None,
    token: str,
) -> dict:
    """Authorize the operator key via Hetzner's ssh_keys API (injected into root),
    allocate a server, wait until docker is up, register a local docker context, and
    return the Host record. On any post-allocation failure, destroys the server AND
    any key we uploaded (no orphaned billable server / dangling key) and dies."""
    user_data = hetzner_build_user_data()
    ctx = docker_context_name(name)
    sock = daemon_socket_path(name)
    automation_pub = generate_automation_key(name)
    server_id: int | None = None
    our_op_key: int | None = None  # operator key — only if WE uploaded it
    auto_key_id: int | None = None  # automation key — always ours, always cleaned up
    # ONE guard around key upload, allocation AND everything after. BaseException
    # (not just Fatal) so a Ctrl-C during the multi-minute wait, a missing docker/
    # ssh binary (OSError), or a decode error still tears everything down (FR-011).
    try:
        op_key_id, created = hetzner_ensure_ssh_key(ctx, operator_pubkey, token)
        our_op_key = op_key_id if created else None
        auto_key_id, _ = hetzner_ensure_ssh_key(f"{ctx}-automation", automation_pub, token)
        # Authorize BOTH on root: the automation key for the tool's unattended docker
        # socket-forward, the operator key for interactive `attach`.
        ssh_keys: list = [op_key_id, auto_key_id] + ([extra_ssh_key] if extra_ssh_key else [])
        resp = hetzner_create_server(name, server_type, location, user_data, token, ssh_keys)
        server = resp.get("server", {})
        sid = server.get("id")
        if not isinstance(sid, int):
            # Redacted: the create response can carry a generated root_password.
            die(f"Hetzner create returned no server id (response keys: {sorted(resp)})")
        server_id = sid
        ip = (server.get("public_net", {}).get("ipv4") or {}).get("ip")
        deadline = time.monotonic() + 120
        while not ip and time.monotonic() < deadline:
            time.sleep(3)
            srv = hetzner_get_server(server_id, token)  # transient GET failures retry internally
            ip = (srv.get("public_net", {}).get("ipv4") or {}).get("ip")
        if not ip:
            die(f"server {server_id} never reported a public IPv4")
        log(f"server {server_id} allocated at {ip}; waiting for docker (cloud-init)…")
        # Context targets the local forwarded socket, not ssh:// — the tool owns the
        # ssh invocation (automation key, no agent) via the _forward_argv tunnel.
        docker_context_create(ctx, f"unix://{sock}")
        wait_until_reachable(ip, ctx)
    except BaseException:
        # A POST may have created a server whose response we never saw — find it by
        # name so it is not orphaned, then delete server + the keys we uploaded, the
        # local automation key, the tunnel and the context.
        if server_id is None:
            found = hetzner_find_server_by_name(name, token)
            server_id = found.get("id") if isinstance(found, dict) else None
        cleanup_on_failure(server_id, token, ssh_key_ids=(our_op_key, auto_key_id), name=name)
        query(["docker", "context", "rm", "-f", ctx])
        raise
    return {
        "driver": "docker",
        "context": ctx,
        "address": ip,
        "provisioning": {
            "provider": "hetzner",
            "server_id": server_id,
            "server_type": server_type,
            "location": location,
            "connection": "ssh-forward",
            "ssh_key_id": our_op_key,  # operator key, only if WE uploaded it (destroy removes it)
            "automation_ssh_key_id": auto_key_id,  # always ours; destroy removes it
            "created": True,
        },
        "created_by_tool": True,
    }


def provisioner_destroy(host: dict, token: str) -> None:
    """Deprovision: delete the server and remove the local docker context. Refuses
    a host this tool did not create (defense in depth for the future US3 caller;
    FR-010 — never destroy infrastructure we did not allocate)."""
    if not host.get("created_by_tool"):
        die("refusing to deprovision a server this tool did not create")
    prov = host.get("provisioning") or {}
    server_id = prov.get("server_id")
    if isinstance(server_id, int):
        # strict: a failed server delete must propagate so the caller keeps the
        # host registered for a retry (never orphan a live billable server). The
        # key/context cleanup below is best-effort and re-runs idempotently.
        hetzner_delete_server(server_id, token, strict=True)
    # Both keys we uploaded; a reused operator key (ssh_key_id is None) is left alone.
    for key_id in (prov.get("ssh_key_id"), prov.get("automation_ssh_key_id")):
        if isinstance(key_id, int):
            hetzner_delete_ssh_key(key_id, token)
    ctx = host.get("context")
    name = host_name_from_context(ctx) if isinstance(ctx, str) else ""
    if name:
        _close_tunnel(name)
        remove_automation_key(name)
    if isinstance(ctx, str) and ctx.startswith(CONTAINER_PREFIX):
        query(["docker", "context", "rm", "-f", ctx])


def provision_host(
    provider: str,
    name: str,
    *,
    server_type: str | None,
    location: str | None,
    ssh_key: str | None,
    ssh_pubkey: Path | None,
) -> dict:
    """Dispatch provisioning to the provider (only 'hetzner' today) and return the
    Host record for host `name`. Token read once from env; billable allocation only."""
    if provider != "hetzner":
        die(f"unknown --provider '{provider}' (only 'hetzner' is supported)")
    if not HETZNER_NAME_RE.fullmatch(name):
        die(
            f"'{name}' is not a valid cloud server name (RFC-1123: lowercase letters, "
            "digits, hyphens; no underscore; <=63 chars)"
        )
    token = _hcloud_token()
    operator_pubkey = resolve_operator_pubkey(ssh_pubkey)
    return provisioner_create(
        name,
        server_type or HETZNER_DEFAULT_SERVER_TYPE,
        location or HETZNER_DEFAULT_LOCATION,
        operator_pubkey,
        ssh_key,
        token,
    )


# --- driver seam (Feature 001) -----------------------------------------------
# A driver abstracts build/run/connect on a host so local and remote share one
# path. Pure argv builders (no runtime needed) — the docker/podman asymmetry is
# absorbed here: docker targets `--context`, podman targets `--connection`. An
# 'existing-ssh' host is attach-only and rejects container operations.


def driver_runtime_argv(host: dict) -> list[str]:
    """Base argv targeting a host's runtime.

    >>> driver_runtime_argv({"driver": "docker", "context": "lima"})
    ['docker', '--context', 'lima']
    >>> driver_runtime_argv({"driver": "podman", "context": "vps"})
    ['podman', '--connection', 'vps']
    >>> driver_runtime_argv({"driver": "docker", "context": ""})
    ['docker']
    """
    driver = host.get("driver")
    ctx = host.get("context") or ""
    if driver == "docker":
        return ["docker", "--context", ctx] if ctx else ["docker"]
    if driver == "podman":
        return ["podman", "--connection", ctx] if ctx else ["podman"]
    die(f"host driver '{driver}' is attach-only; it cannot build or run containers")


# --- sidecar / helper services (Feature 002 US4, research R5) -----------------
# A deployment may declare helper services in an operator-supplied compose
# OVERRIDE file, discovered by convention next to the .env. When present it is
# merged as a SECOND `-f` into every compose invocation for that deployment, so
# the agent and its helpers share one project and one lifecycle (FR-004). This is
# a deliberately thin, file-based seam — the richer whole-directory model is
# Feature 006 (agent-as-code), which builds ON this primitive.

# The tool owns the agent service's identity: its service key, plus the top-level
# project name / named volumes / injected configs. An override MUST NOT redefine
# any of these — it may only ADD helper services (services-only fragment).
AGENT_SERVICE_KEY = "agent"
SIDECAR_ALLOWED_TOPLEVEL = {"services", "version"}  # version: tolerated (compose-deprecated)


def sidecar_override_candidates(name: str, cwd: Path) -> list[Path]:
    """Discovery order, mirroring .env resolution (project-local first, then the
    per-name user config).

    >>> [p.name for p in sidecar_override_candidates("acme", Path("/w"))]
    ['agent-container.acme.services.yaml', 'acme.services.yaml']
    """
    return [cwd / f"agent-container.{name}.services.yaml", CONFIG_DIR / f"{name}.services.yaml"]


def resolve_sidecar_override(name: str) -> Path | None:
    """The first existing sidecar override for <name>, validated, or None. A
    present-but-invalid override is fatal (FR-018) — never silently ignored."""
    for candidate in sidecar_override_candidates(name, Path.cwd()):
        if candidate.is_file():
            validate_sidecar_override(candidate)
            return candidate
    return None


def _yaml_toplevel_keys(text: str) -> list[str]:
    """Top-level mapping keys of a YAML/JSON document, best-effort. JSON is a YAML
    subset, so a JSON override parses exactly; a YAML override is scanned for
    column-0 `key:` lines (comments / list items / nested lines ignored). This is
    a GUARDRAIL, not a full parser — compose does the authoritative parse when it
    runs; the scan only catches the two footguns (non-services-only; agent redef)."""
    try:
        obj = json.loads(text)
        return [str(k) for k in obj] if isinstance(obj, dict) else []
    except json.JSONDecodeError, ValueError:
        pass
    keys = []
    for line in text.splitlines():
        if not line or line[0].isspace() or line.lstrip().startswith("#") or line.startswith("---"):
            continue
        m = re.match(r"([A-Za-z0-9_.-]+)\s*:", line)
        if m:
            keys.append(m.group(1))
    return keys


def _yaml_service_keys(text: str) -> list[str]:
    """Service keys declared under a top-level `services:` mapping, best-effort
    (JSON parsed exactly; YAML scanned at the first indent level under services)."""
    try:
        obj = json.loads(text)
        svc = obj.get("services") if isinstance(obj, dict) else None
        return [str(k) for k in svc] if isinstance(svc, dict) else []
    except json.JSONDecodeError, ValueError:
        pass
    lines = text.splitlines()
    keys: list[str] = []
    indent: int | None = None
    in_services = False
    for line in lines:
        if re.match(r"services\s*:", line):  # column-0 services:
            in_services = True
            continue
        if in_services:
            if line and not line[0].isspace() and not line.lstrip().startswith("#"):
                break  # dedented back to another top-level key
            stripped = line.strip()
            if not stripped or stripped.startswith("#"):
                continue
            cur = len(line) - len(line.lstrip())
            if indent is None:
                indent = cur
            if cur == indent:
                m = re.match(r"([A-Za-z0-9_.-]+)\s*:", stripped)
                if m:
                    keys.append(m.group(1))
    return keys


def validate_sidecar_override(path: Path) -> None:
    """Guardrail on an operator sidecar override (FR-004/FR-018): it must be a
    non-empty compose `services:`-only fragment and MUST NOT redefine the agent
    service (whose identity — name/port/the 7 volumes — the tool owns). Deeper
    malformations are left to compose's own parse at run time; here we reject the
    two footguns with an actionable message rather than silently merging them."""
    try:
        text = path.read_text()
    except OSError as e:
        die(f"sidecar override {path} is unreadable: {e}")
    if not text.strip():
        die(f"sidecar override {path} is empty (declare helper services under `services:`)")
    toplevel = _yaml_toplevel_keys(text)
    if "services" not in toplevel:
        die(f"sidecar override {path} has no `services:` block (it must add helper services)")
    extra = [k for k in toplevel if k not in SIDECAR_ALLOWED_TOPLEVEL]
    if extra:
        die(
            f"sidecar override {path} may only declare `services:` (a services-only "
            f"fragment); remove tool-owned/unsupported top-level key(s): {', '.join(sorted(extra))}"
        )
    if AGENT_SERVICE_KEY in _yaml_service_keys(text):
        die(
            f"sidecar override {path} must not redefine the '{AGENT_SERVICE_KEY}' service "
            f"(the tool owns its name, port, and volumes); name your helper differently"
        )


def driver_compose_argv(
    host: dict, project: str, file: Path, *args: str, override: Path | None = None
) -> list[str]:
    """`<runtime> --context/-connection X compose -p <project> -f <file> [-f <override>] …`.

    When a sidecar `override` is given it rides as a SECOND `-f` right after the
    generated compose file, so compose's native multi-file merge folds the helper
    services into the same project (US4). Order matters: the override is layered
    ON TOP of the generated file.

    >>> from pathlib import Path
    >>> driver_compose_argv({"driver": "docker", "context": "lima"}, "agent-container-acme", Path("/s/acme.compose.yaml"), "up", "-d")
    ['docker', '--context', 'lima', 'compose', '-p', 'agent-container-acme', '-f', '/s/acme.compose.yaml', 'up', '-d']
    >>> driver_compose_argv({"driver": "docker", "context": "lima"}, "p", Path("/s/a.yaml"), "up", override=Path("/s/side.yaml"))
    ['docker', '--context', 'lima', 'compose', '-p', 'p', '-f', '/s/a.yaml', '-f', '/s/side.yaml', 'up']
    """
    files = ["-f", str(file)] + (["-f", str(override)] if override else [])
    return driver_runtime_argv(host) + ["compose", "-p", project, *files, *args]


def driver_up_argv(host: dict, project: str, file: Path, override: Path | None = None) -> list[str]:
    return driver_compose_argv(host, project, file, "up", "-d", "--build", override=override)


def driver_redeploy_argv(
    host: dict, project: str, file: Path, override: Path | None = None
) -> list[str]:
    """Deliberately non-idempotent (FR-010): rebuild the image and recreate the
    container even with no change. Named volumes are external-by-name so they are
    re-attached, not recreated (FR-008)."""
    return driver_compose_argv(
        host, project, file, "up", "-d", "--build", "--force-recreate", override=override
    )


def driver_stop_argv(
    host: dict, project: str, file: Path, override: Path | None = None
) -> list[str]:
    return driver_compose_argv(host, project, file, "stop", override=override)


def driver_start_argv(
    host: dict, project: str, file: Path, override: Path | None = None
) -> list[str]:
    return driver_compose_argv(host, project, file, "start", override=override)


def driver_down_argv(
    host: dict,
    project: str,
    file: Path,
    purge: bool = False,
    rmi_local: bool = False,
    override: Path | None = None,
) -> list[str]:
    args = ["down"]
    if purge:
        args.append("--volumes")
    if rmi_local:  # wipe also removes the locally-built image (never a public one)
        args += ["--rmi", "local"]
    return driver_compose_argv(host, project, file, *args, override=override)


def driver_reachable_address(host: dict) -> str:
    """Address used for attach: the host's own address (localhost for local)."""
    return host.get("address") or "localhost"


# --- compose generation (Feature 001) ----------------------------------------
# The per-container deployment is a generated, inspectable compose project,
# emitted as JSON (a valid YAML subset) so no YAML dependency is needed. It
# declares the seven named volumes and expresses injected SSH identity as compose
# `secrets` (private host key) / `configs` (public authorized_keys) referencing
# LOCAL files, so the material transfers over a remote context (a bind would
# resolve empty on the remote). No secret value is ever written inline — only
# `file:` references. Regenerated from parameters on every up (derived artifact).


def compose_project(name: str) -> str:
    """
    >>> compose_project("acme")
    'agent-container-acme'
    """
    return container_name(name)


def build_compose_model(
    name: str,
    build_context: Path | str,
    host_key_file: Path | None = None,
    authorized_keys_file: Path | None = None,
    env_file: Path | None = None,
    extra_mounts: list[str] | None = None,
) -> dict:
    """Build the compose model for container <name>. Pure: embeds paths as
    strings, reads nothing. host_key_file/authorized_keys_file are LOCAL staged
    files referenced via compose secrets/configs (never inlined). env_file is
    read client-side by compose and merged into the service environment;
    extra_mounts are already-resolved '<abs-host>:<container>' bind specs.

    NOTE: the exact in-container mount path for compose secrets vs configs (an
    absolute `target`) is validated in the acceptance tier (research R5); the
    entrypoint reads the INJECT_* paths.
    """
    port = port_for_name(name)
    service: dict = {
        "container_name": container_name(name),
        "build": {"context": str(build_context)},
        "restart": "unless-stopped",
        "ports": [f"{port}:2222"],
        "volumes": list(all_volume_mounts(name)) + list(extra_mounts or []),
    }
    if env_file is not None:
        service["env_file"] = [str(env_file)]
    model: dict = {
        "name": compose_project(name),
        "services": {"agent": service},
        # Pin each volume's `name` so compose does NOT prefix it with the project
        # ("<project>_<vol>"); the seven volume names are the deterministic identity
        # contract (Constitution IV) that `--purge` and the completions rely on.
        "volumes": {vn: {"name": vn} for vn in per_container_volumes(name)},
    }
    # Injected SSH identity rides as compose `configs` (NOT `secrets`). Both the
    # host key and authorized_keys are delivered this way because a compose
    # `secret` with an ABSOLUTE target crash-loops the container on some docker
    # engines (observed on GitHub ubuntu runners; compose `configs` with an
    # absolute target are portable). The material is still handled as sensitive:
    # staged locally 0600, ephemeral (never persisted to a volume), never baked
    # into the image, never on argv (Constitution III — least exposure holds; the
    # secret-vs-config choice is a portability-forced compose mechanism detail).
    svc_configs: list[dict] = []
    model_configs: dict = {}
    if host_key_file is not None:
        svc_configs.append({"source": "ssh_host_key", "target": INJECT_HOST_KEY_PATH})
        model_configs["ssh_host_key"] = {"file": str(host_key_file)}
    if authorized_keys_file is not None:
        svc_configs.append({"source": "ssh_authorized_keys", "target": INJECT_AUTHORIZED_KEYS_PATH})
        model_configs["ssh_authorized_keys"] = {"file": str(authorized_keys_file)}
    if svc_configs:
        service["configs"] = svc_configs
        model["configs"] = model_configs
    return model


def write_compose_file(host: str, name: str, model: dict) -> Path:
    """Write the (regenerable) compose model as JSON to the per-host state dir."""
    host_state_dir(host).mkdir(parents=True, exist_ok=True)
    p = compose_file_path(host, name)
    p.write_text(json.dumps(model, indent=2) + "\n")
    return p


def quadlet_active(name: str) -> bool:
    # Best-effort probe; systemctl is absent on macOS — ignore all failures.
    if not shutil.which("systemctl"):
        return False
    try:
        r = query(["systemctl", "--user", "is-active", f"{container_name(name)}.service"])
        return r.returncode == 0 and r.stdout.strip() == "active"
    except Exception:
        return False


def is_tty() -> bool:
    return sys.stdin.isatty() and sys.stdout.isatty()


# --- commands ----------------------------------------------------------------


def log(msg: str) -> None:
    eprint(f"[agent-container] {msg}")


def warn(msg: str) -> None:
    eprint(f"[agent-container] WARNING: {msg}")


def hint(cmd: str) -> None:
    console.print(f"hint: {cmd}", style="dim")


def do_build(tag: str, context: Path | None = None) -> None:
    # `build` is the one subcommand that needs a repo checkout: a docker build
    # context IS a checkout. Resolve it explicitly (a PyPI install has REPO_ROOT
    # is None) and fail actionably rather than with a traceback.
    ctx = context
    if ctx is None:
        env = os.environ.get("AGENT_CONTAINER_REPO")
        if env:
            base = Path(env).expanduser().resolve()
            if not _is_repo_checkout(base):
                die(
                    f"AGENT_CONTAINER_REPO={env} is not an agent-container checkout "
                    "(missing Dockerfile/completions/agent-container.bash)"
                )
            ctx = base
        else:
            ctx = REPO_ROOT
    if ctx is None:
        die(
            "no repo checkout for the docker build context; run from a checkout, "
            "set AGENT_CONTAINER_REPO=<checkout>, or pass --context <dir>"
        )
    rt = detect_runtime()
    log(f"building image '{tag}' with {rt} from {ctx}")
    # Inherited stdio: layer progress must stream, never capture.
    rc = subprocess.run([rt, "build", "-t", tag, str(ctx)]).returncode
    if rc != 0:
        die(f"build failed (exit {rc})")


def resolve_build_context() -> Path:
    """The repo checkout used as the compose build context (image builds on the
    host). Same resolution as do_build (AGENT_CONTAINER_REPO -> REPO_ROOT); dies
    actionably when no checkout is reachable (e.g. a bare PyPI install)."""
    env = os.environ.get("AGENT_CONTAINER_REPO")
    if env:
        base = Path(env).expanduser().resolve()
        if not _is_repo_checkout(base):
            die(f"AGENT_CONTAINER_REPO={env} is not an agent-container checkout")
        return base
    if REPO_ROOT is not None:
        return REPO_ROOT
    die(
        "no repo checkout for the compose build context; run from a checkout "
        "or set AGENT_CONTAINER_REPO=<checkout>"
    )


def host_is_local(host: dict) -> bool:
    return driver_reachable_address(host) == "localhost"


def implicit_local_host(rt: str | None = None) -> dict:
    """A synthetic 'local' host using the default runtime context — the smooth
    upgrade path from the pre-registry single-host tool."""
    return {
        "driver": rt or detect_runtime(),
        "context": "",
        "address": "localhost",
        "provisioning": None,
        "created_by_tool": False,
    }


def resolve_deploy_host(host_name: str | None) -> tuple[str, dict]:
    """Resolve (host_name, host_record) for a deploy/lifecycle op. An explicit
    --host must exist; otherwise the registry default; and if the registry is
    empty, an implicit local host (behaves like the pre-multi-host tool)."""
    reg = load_registry()
    if host_name is not None:
        h = get_host(reg, host_name)
        if h is None:
            die(f"no host named '{host_name}' (see: agent-container host ls)")
        return host_name, h
    default = default_host_name(reg)
    if default is not None:
        h = get_host(reg, default)
        if h is not None:
            return default, h
    log(
        f"no hosts registered; using an implicit local host "
        f"(register one with: agent-container host add {DEFAULT_HOST} --docker-context <ctx>)"
    )
    return DEFAULT_HOST, implicit_local_host()


def host_ps_rows(host: dict, include_stopped: bool = False) -> list[tuple[str, str, str, str]]:
    """(cname, image, status, uptime) rows for agent-container-* on a host's daemon.
    Fail-CLOSED (001-US3 lesson): an unreachable/timed-out/errored `ps` RAISES
    (Fatal via ensure_tunnel(required)/returncode, or subprocess.TimeoutExpired)
    rather than returning [] — a down daemon must never read as 'no containers'.
    A reachable host returns 0, so lifecycle existence-checks are unaffected."""
    ensure_tunnel(host, required=True)
    argv = driver_runtime_argv(host) + ["ps"] + (["-a"] if include_stopped else [])
    argv += ["--format", "{{.Names}}\t{{.Image}}\t{{.Status}}\t{{.RunningFor}}"]
    try:
        r = query(argv, timeout=20)  # bounded so a wedged forward can't hang the caller
    except (subprocess.SubprocessError, OSError) as e:
        # A wedged-but-bound forward times out here; a spawn failure raises OSError.
        # Convert to Fatal so lifecycle callers (do_up/down/redeploy) fail-closed
        # via the normal die() path instead of an uncaught TimeoutExpired traceback.
        die(f"could not list containers on the host: {e}")
    if r.returncode != 0:
        die(f"could not list containers on the host: {r.stderr.strip() or f'exit {r.returncode}'}")
    rows = []
    for line in r.stdout.splitlines():
        parts = line.split("\t")
        if len(parts) == 4 and parts[0].startswith(CONTAINER_PREFIX):
            rows.append((parts[0], parts[1], parts[2], parts[3]))
    return rows


def host_container_names(host: dict, include_stopped: bool = False) -> set[str]:
    return {row[0] for row in host_ps_rows(host, include_stopped)}


def assert_host_empty(host: dict) -> None:
    """Fail-CLOSED container check for `host rm --destroy` (FR-009 / SC-005):
    permit deprovisioning ONLY when a container listing that PROVABLY succeeded
    reports zero agent-container-* containers. `host_ps_rows`/`query` are
    best-effort (fine for `list`), but on the teardown path a failed/unreachable
    `docker ps` returns empty stdout that is indistinguishable from a genuinely
    empty daemon — so trusting it would let a loaded (or unreachable) server be
    destroyed. Here any enumeration failure is fatal. include_stopped: a stopped
    agent-container-* still occupies the server and holds its volumes."""
    ensure_tunnel(host, required=True)  # a down forward must refuse, not read empty
    argv = driver_runtime_argv(host) + ["ps", "-a", "--format", "{{.Names}}"]
    r = query(argv)
    if r.returncode != 0:
        die(
            "refusing to destroy: could not confirm the host is empty "
            f"(listing containers failed: {r.stderr.strip() or f'exit {r.returncode}'})"
        )
    names = sorted(n for n in r.stdout.splitlines() if n.startswith(CONTAINER_PREFIX))
    if names:
        hostname = host_name_from_context(host.get("context") or "")
        die(
            f"refusing to destroy: {len(names)} container(s) still present "
            f"({', '.join(names)}); remove them first "
            f"(e.g. agent-container down <name> --host {hostname})"
        )


def stage_ssh_injection(
    host: str, name: str, host_key: Path | None, authorized_keys: list[Path]
) -> tuple[Path | None, Path | None]:
    """Stage injected SSH material as LOCAL files under the per-host state dir,
    returning (host_key_file, authorized_keys_file) for the compose model to
    reference as configs (transfers over a remote context; a bind resolves empty
    on a remote host).

    Both files are staged **0644**: compose exposes the source file's mode into
    the container, where the unprivileged `dev` user reads it — and dev's uid
    (1000) need NOT equal the host uid that ran `up` (it does not on CI runners),
    so a 0600 file is unreadable inside the container and crash-loops the
    entrypoint. 0644 is safe because the per-host state dir is locked to 0700, so
    no other HOST user can traverse to the files, while the container reads each
    inode through the mount. (This mirrors the original bind-mount rationale;
    staging 0600 here was a regression that broke on ubuntu docker.)"""
    d = host_state_dir(host)
    d.mkdir(parents=True, exist_ok=True)
    d.chmod(0o700)  # host-side protection for the 0644 key files
    hk_file: Path | None = None
    ak_file: Path | None = None
    if host_key is not None:
        hk = host_key.expanduser()
        if not hk.is_file():
            die(f"--host-key: {host_key} does not exist or is not a file")
        validate_private_key(hk)
        hk_file = d / f"{name}.host_key"
        hk_file.write_bytes(hk.read_bytes())
        hk_file.chmod(0o644)
    if authorized_keys:
        blob = ""
        for p in authorized_keys:
            pk = p.expanduser()
            if not pk.is_file():
                die(f"--authorized-key: {p} does not exist or is not a file")
            text = pk.read_text()
            blob += text if text.endswith("\n") else text + "\n"
        ak_file = d / f"{name}.authorized_keys"
        ak_file.write_text(blob)
        ak_file.chmod(0o644)
    return hk_file, ak_file


def compose_up_exec(
    host_name: str,
    host_rec: dict,
    name: str,
    env_file: Path,
    mounts: list[str],
    host_key: Path | None,
    authorized_keys: list[Path],
    redeploy: bool = False,
) -> None:
    """Generate the compose project and bring it up on the host (image builds on
    the host). With redeploy=True it force-recreates (US2, deliberately
    non-idempotent) — the container already holds its port, so the port-free
    guard is skipped and its named volumes are preserved across the recreate."""
    port = port_for_name(name)
    if not redeploy and host_is_local(host_rec) and not port_free(port):
        die(
            f"port {port} is already in use (port = 2200 + name hash; "
            f"collisions are possible). Stop whatever holds it (agent-container list) "
            f"or pick a different name."
        )
    hk_file, ak_file = stage_ssh_injection(host_name, name, host_key, authorized_keys)
    build_ctx = resolve_build_context()
    model = build_compose_model(name, build_ctx, hk_file, ak_file, env_file, mounts)
    compose_file = write_compose_file(host_name, name, model)
    log(f"host={host_name} name={name} port={port} env-file={env_file}")
    log(f"compose={compose_file}")
    project = compose_project(name)
    override = resolve_sidecar_override(name)  # validated; merged as a 2nd -f (US4)
    if override is not None:
        log(f"sidecar override={override}")
    argv = (
        driver_redeploy_argv(host_rec, project, compose_file, override)
        if redeploy
        else driver_up_argv(host_rec, project, compose_file, override)
    )
    rc = subprocess.run(argv).returncode  # inherited stdio: build output streams
    if rc != 0:
        die(f"compose {'redeploy' if redeploy else 'up'} failed (exit {rc})")
    write_state(host_name, name, port)
    addr = driver_reachable_address(host_rec)
    hflag = "" if host_name == DEFAULT_HOST else f" --host {host_name}"
    verb = "redeployed" if redeploy else "started"
    log(f"{verb} {container_name(name)}; attach with: agent-container attach {name}{hflag}")
    log(f"or directly: ssh dev@{addr} -p {port} -t tmux attach -t main")


def do_up(
    name: str,
    host: str | None = None,
    env_file_override: Path | None = None,
    mounts: list[str] | None = None,
    host_key: Path | None = None,
    authorized_keys: list[Path] | None = None,
) -> None:
    validate_name(name)
    migrate_flat_state()
    # Resolve binds up front so a bad --mount aborts before any runtime call.
    resolved_mounts = [resolve_bind_mount(m) for m in (mounts or [])]
    host_name, host_rec = resolve_deploy_host(host)
    ensure_tunnel(host_rec)
    cname = container_name(name)
    hflag = "" if host_name == DEFAULT_HOST else f" --host {host_name}"

    existing = host_container_names(host_rec, include_stopped=True)
    if cname in host_container_names(host_rec):
        port = read_state_port(host_name, name)
        log(
            f"container {cname} already running on {host_name}"
            + (f" (port {port})" if port else "")
        )
        log(f"attach with: agent-container attach {name}{hflag}")
        return
    if cname in existing:
        die(
            f"container {cname} exists but is not running on {host_name}. Run: agent-container down {name}{hflag}"
        )

    env_file = env_file_override or resolve_env_file(name)
    if env_file is None:
        cands = ", ".join(str(c) for c in env_file_candidates(name, Path.cwd()))
        die(f"no .env found. Looked in: {cands}")

    # The image builds on the host via compose (`up --build`) — no pre-built
    # image or registry needed. Serialized against concurrent lifecycle ops (FR-017).
    with deployment_lock(host_name, name):
        compose_up_exec(
            host_name, host_rec, name, env_file, resolved_mounts, host_key, authorized_keys or []
        )


def down_container(
    host_name: str, host_rec: dict, name: str, purge: bool, rmi_local: bool = False
) -> None:
    ensure_tunnel(host_rec)
    cname = container_name(name)
    compose_file = compose_file_path(host_name, name)
    # host_container_names is fail-closed (it dies on an unreachable daemon). On
    # the teardown path that would strand the local state: down/wipe must still
    # clear the per-host state (and drop the derived compose artifact on purge)
    # even when the host is gone, so re-adding it later starts clean. Degrade a
    # failed enumeration to 'unknown' and fall through to the compose-file path.
    try:
        exists = cname in host_container_names(host_rec, include_stopped=True)
    except (Fatal, OSError, subprocess.SubprocessError) as e:
        warn(f"could not confirm container state on {host_name} ({e}); clearing local state anyway")
        exists = False
    vols = per_container_volumes(name)

    if exists or compose_file.is_file():
        log(f"stopping and removing {cname} on {host_name}")
        if compose_file.is_file():
            # Merge the sidecar override (US4) so `down`/`wipe` tear the helpers down
            # as one unit — but a now-invalid/removed override must never BLOCK a
            # teardown (compose down also reconciles by project label), so resolve it
            # leniently here, unlike the fail-fast create path.
            try:
                override = resolve_sidecar_override(name)
            except Fatal as e:
                warn(f"ignoring sidecar override for teardown ({e})")
                override = None
            # compose down removes the container (+ networks); --volumes also drops
            # the seven named volumes; --rmi local also removes the locally-built
            # image (wipe). Without those, volumes/image are preserved.
            query(
                driver_down_argv(
                    host_rec, compose_project(name), compose_file, purge, rmi_local, override
                )
            )
        else:
            # No compose file (e.g. a pre-compose container): fall back to rm and,
            # on purge, explicit volume removal.
            query(driver_runtime_argv(host_rec) + ["rm", "-f", cname])
            if purge:
                for vname in vols:
                    query(driver_runtime_argv(host_rec) + ["volume", "rm", vname])
        # `rm`/compose-down returns before the daemon frees the published port;
        # wait so an immediate re-`up` on the same name doesn't hit 'port in use'.
        if host_is_local(host_rec):
            wait_port_released(port_for_name(name))
    else:
        log(f"no container named {cname} on {host_name}")

    if purge:
        compose_file.unlink(missing_ok=True)  # the derived artifact goes too
        log(f"purged volumes: {', '.join(vols)}")
    else:
        log(f"volumes preserved (use --purge to remove): {', '.join(vols)}")

    clear_state(host_name, name)


# Host-side paths of the persisted SSH material inside the container (rootless,
# dev-owned). Kept in sync with entrypoint.sh / the Dockerfile HostKey directive.
CONTAINER_HOSTKEY = "/home/dev/.ssh/hostkeys/ssh_host_ed25519_key"
CONTAINER_AUTHKEYS = "/home/dev/.ssh/authorized_keys"


def inject_keys(rt: str, name: str, host_key: Path | None, authorized_keys: list[Path]) -> None:
    """Inject an SSH host key and/or authorized_keys into a RUNNING container,
    onto the persisted ~/.ssh volume, without recreating it. The host key is
    copied in and installed (then sshd is HUP-reloaded); public keys are
    streamed over stdin (never on argv) and merged with dedup. Rootless: every
    in-container step runs as the dev user, no sudo."""
    cname = container_name(name)
    if host_key is not None:
        hk = host_key.expanduser()
        if not hk.is_file():
            die(f"--host-key: {host_key} does not exist or is not a file")
        validate_private_key(hk)
        # Stream the key over stdin (not `cp`): `cp` lands it owned by the host
        # uid, unreadable by the rootless `dev` user. Writing via `cat` inside
        # the container gives dev ownership, and the key never touches argv.
        install_sh = (
            f"umask 077; mkdir -p /home/dev/.ssh/hostkeys; "
            f"cat > {CONTAINER_HOSTKEY}; chmod 600 {CONTAINER_HOSTKEY}; "
            f"ssh-keygen -y -f {CONTAINER_HOSTKEY} > {CONTAINER_HOSTKEY}.pub && "
            f"pkill -HUP -x sshd"
        )
        r = subprocess.run(
            [rt, "exec", "-i", cname, "bash", "-lc", install_sh], input=hk.read_bytes()
        )
        if r.returncode != 0:
            die(f"failed to install host key in {cname}")
        log(f"installed host key into {cname} and reloaded sshd")
    for p in authorized_keys:
        pk = p.expanduser()
        if not pk.is_file():
            die(f"--authorized-key: {p} does not exist or is not a file")
        append_sh = (
            f"umask 077; mkdir -p /home/dev/.ssh; cat >> {CONTAINER_AUTHKEYS}; "
            f"awk 'NF && !seen[$0]++' {CONTAINER_AUTHKEYS} > {CONTAINER_AUTHKEYS}.tmp && "
            f"mv {CONTAINER_AUTHKEYS}.tmp {CONTAINER_AUTHKEYS}; chmod 600 {CONTAINER_AUTHKEYS}"
        )
        r = subprocess.run(
            [rt, "exec", "-i", cname, "bash", "-lc", append_sh], input=pk.read_bytes()
        )
        if r.returncode != 0:
            die(f"failed to inject authorized key {p}")
        log(f"injected authorized key {pk.name} into {cname}")


def cli_keys(name: str, host_key: Path | None, authorized_keys: list[Path]) -> None:
    validate_name(name)
    if host_key is None and not authorized_keys:
        die("keys: nothing to inject — pass --host-key and/or --authorized-key")
    rt = detect_runtime()
    cname = container_name(name)
    if not container_running(rt, cname):
        die(f"container {cname} is not running. Start it first: agent-container up {name}")
    inject_keys(rt, name, host_key, authorized_keys)


def cli_down(name: str, purge: bool, yes: bool, host: str | None = None) -> None:
    validate_name(name)
    host_name, host_rec = resolve_deploy_host(host)

    if host_is_local(host_rec) and quadlet_active(name):
        warn(
            f"systemd user unit {container_name(name)}.service is active; "
            f"systemd will restart the container after removal. Manage it via Quadlet instead."
        )
        if not yes:
            eprint("[agent-container] re-run with -y/--yes to proceed despite the active unit")
            raise typer.Exit(2)

    if not yes:
        if not is_tty():
            eprint("[agent-container] refusing destructive 'down' without -y/--yes on a non-TTY")
            raise typer.Exit(2)
        what = f"stop and remove {container_name(name)} on {host_name}"
        if purge:
            what += (
                f" and DELETE all per-container volumes ({', '.join(per_container_volumes(name))})"
            )
        if not questionary.confirm(f"{what}?", default=False).ask():
            log("aborted")
            return

    with deployment_lock(host_name, name):  # serialize vs concurrent lifecycle ops (FR-017)
        down_container(host_name, host_rec, name, purge)


def do_stop(name: str, host: str | None = None) -> None:
    """Pause/reclaim (FR-006): halt the deployment (all services) keeping the
    container and its volumes; `start` resumes it without recreation."""
    validate_name(name)
    host_name, host_rec = resolve_deploy_host(host)
    compose_file = compose_file_path(host_name, name)
    if not compose_file.is_file():
        die(
            f"no deployment named '{name}' on {host_name} to stop (deploy it: agent-container up {name})"
        )
    override = resolve_sidecar_override(name)  # act on the agent + helpers as one unit (US4)
    with deployment_lock(host_name, name):
        ensure_tunnel(host_rec)
        argv = driver_stop_argv(host_rec, compose_project(name), compose_file, override)
        if subprocess.run(argv).returncode != 0:
            die("compose stop failed")
    log(f"stopped {container_name(name)} on {host_name} (resume: agent-container start {name})")


def do_start(name: str, host: str | None = None) -> None:
    """Resume a stopped deployment (FR-006): no recreation, no rebuild."""
    validate_name(name)
    host_name, host_rec = resolve_deploy_host(host)
    compose_file = compose_file_path(host_name, name)
    if not compose_file.is_file():
        die(f"no deployment named '{name}' on {host_name} (deploy it: agent-container up {name})")
    override = resolve_sidecar_override(name)  # act on the agent + helpers as one unit (US4)
    with deployment_lock(host_name, name):
        ensure_tunnel(host_rec)
        argv = driver_start_argv(host_rec, compose_project(name), compose_file, override)
        if subprocess.run(argv).returncode != 0:
            die(
                f"compose start failed — if the deployment was disposed, run: agent-container up {name}"
            )
    hflag = "" if host_name == DEFAULT_HOST else f" --host {host_name}"
    log(
        f"started {container_name(name)} on {host_name}; attach with: agent-container attach {name}{hflag}"
    )


def do_redeploy(
    name: str,
    host: str | None = None,
    env_file_override: Path | None = None,
    mounts: list[str] | None = None,
    host_key: Path | None = None,
    authorized_keys: list[Path] | None = None,
) -> None:
    """Redeploy (FR-008/FR-010): rebuild the image on the host and recreate the
    container, preserving its volumes. Deliberately NON-idempotent — always
    recreates even with no change, since the operator explicitly asked to rebuild."""
    validate_name(name)
    migrate_flat_state()
    resolved_mounts = [resolve_bind_mount(m) for m in (mounts or [])]
    host_name, host_rec = resolve_deploy_host(host)
    ensure_tunnel(host_rec)
    if container_name(name) not in host_container_names(host_rec, include_stopped=True):
        warn(f"no existing container '{name}' on {host_name} — redeploy will create it fresh")
    env_file = env_file_override or resolve_env_file(name)
    if env_file is None:
        cands = ", ".join(str(c) for c in env_file_candidates(name, Path.cwd()))
        die(f"no .env found. Looked in: {cands}")
    with deployment_lock(host_name, name):
        compose_up_exec(
            host_name,
            host_rec,
            name,
            env_file,
            resolved_mounts,
            host_key,
            authorized_keys or [],
            redeploy=True,
        )


def do_wipe(name: str, yes: bool, host: str | None = None) -> None:
    """Wipe (FR-009): remove the container, its persistent volumes, AND its
    locally-built image. Requires explicit confirmation (destroys durable state)."""
    validate_name(name)
    host_name, host_rec = resolve_deploy_host(host)
    if not yes:
        if not is_tty():
            eprint("[agent-container] refusing destructive 'wipe' without -y/--yes on a non-TTY")
            raise typer.Exit(2)
        vols = ", ".join(per_container_volumes(name))
        prompt = (
            f"WIPE {container_name(name)} on {host_name} — delete the container, its "
            f"volumes ({vols}), and its locally-built image?"
        )
        if not questionary.confirm(prompt, default=False).ask():
            log("aborted")
            return
    with deployment_lock(host_name, name):
        down_container(host_name, host_rec, name, purge=True, rmi_local=True)
    log(f"wiped {container_name(name)} on {host_name}")


def gather_rows(rt: str, local_only: bool = False) -> list[dict[str, object]]:
    """Rows for `list`, reconciled against LIVE host state (Feature 002 US3,
    FR-011/012). Local status comes from the local runtime; each REGISTERED remote
    host is queried live via host_ps_rows so status is truthful after an
    out-of-band change (SC-004). A host whose enumeration fails is rendered
    'unreachable' (kept, never shown running, never hangs — bounded + caught), not
    dropped. `local_only` skips all remote round-trips for a fast local-only view."""
    rows: list[dict[str, object]] = []
    seen: set[tuple[str, str]] = set()  # (host, short) — a live row supersedes its placeholder
    reconciled: set[str] = set()  # hosts whose live ps PROVABLY succeeded (orphans -> stale)
    unreachable: set[str] = set()  # hosts whose live ps failed (kept, shown 'unreachable')
    hosts = registry_hosts(load_registry())  # one read; reused for the alias computation below
    # A registered host that actually points at the local daemon (host_is_local) shares
    # the local runtime, so its containers already appear in the local ps below. Query it
    # once (as local), not again as a remote, and skip its state dir so it doesn't render
    # a duplicate 'on remote host' placeholder for a container that is really live locally.
    local_aliases = {h for h, hr in hosts.items() if h != DEFAULT_HOST and host_is_local(hr)}

    # Local host: FAIL-CLOSED (001-US3 lesson) — a failed local `ps` must not read as
    # 'zero containers'; render the local host 'unreachable' instead. include_stopped so a
    # locally stopped container reports its real Exited status, matching the remote path.
    try:
        local_ps = ps_agent_container(rt, include_stopped=True, strict=True)
        reconciled.add(DEFAULT_HOST)
    except Fatal, OSError, subprocess.SubprocessError:
        local_ps = []
        unreachable.add(DEFAULT_HOST)
    for cname, image, status, uptime in local_ps:
        short = cname[len(CONTAINER_PREFIX) :]
        seen.add((DEFAULT_HOST, short))
        rows.append(
            {
                "name": cname,
                "host": DEFAULT_HOST,
                "port": read_state_port(DEFAULT_HOST, short) or "?",
                "image": image,
                "status": status,
                "uptime": uptime,
                "stale": False,
            }  # fmt: skip
        )
    # Live-reconcile every registered REMOTE host (the local host is already covered
    # by ps above). One dead host must never abort or stall the whole listing.
    if not local_only:
        for hname, hrec in sorted(hosts.items()):
            if hname == DEFAULT_HOST or hname in local_aliases:
                continue
            if hrec.get("driver") not in ("docker", "podman"):
                continue  # attach-only/existing-ssh: reachable via ssh, not queryable
            try:
                live = host_ps_rows(
                    hrec, include_stopped=True
                )  # incl. stopped -> real Exited (SC-004)
                reconciled.add(hname)
            except Fatal, OSError, subprocess.SubprocessError:
                unreachable.add(hname)
                continue
            for cname, image, status, uptime in live:
                short = cname[len(CONTAINER_PREFIX) :]
                seen.add((hname, short))
                rows.append(
                    {
                        "name": container_name(short),
                        "host": hname,
                        "port": read_state_port(hname, short) or "?",
                        "image": image,
                        "status": status,
                        "uptime": uptime,
                        "stale": False,
                    }  # fmt: skip
                )
    # Per-host state files (orphans + placeholders for hosts not live-reconciled).
    listed_hosts: set[str] = set()
    if STATE_DIR.is_dir():
        for host_dir in sorted(p for p in STATE_DIR.iterdir() if p.is_dir()):
            hname = host_dir.name
            if hname in local_aliases:
                continue  # its live state is already shown under the local host
            for f in sorted(host_dir.glob("*.port")):
                short = f.stem
                if (hname, short) in seen:  # a live row already supersedes this placeholder
                    continue
                # A state file with no live match on a host we PROVABLY queried means the
                # container is gone (stale); on an unreachable host we can't tell (keep,
                # 'unreachable'); on a host we never queried (--local, attach-only) it's a
                # static 'on remote host' placeholder.
                if hname in unreachable:
                    status, stale = "unreachable", False
                elif hname in reconciled:
                    status, stale = "stale", True
                else:
                    status, stale = "on remote host", False
                listed_hosts.add(hname)
                rows.append(
                    {
                        "name": container_name(short),
                        "host": hname,
                        "port": read_state_port(hname, short) or "?",
                        "image": "-",
                        "status": status,
                        "uptime": "-",
                        "stale": stale,
                    }  # fmt: skip
                )
    # Never let an unreachable host silently vanish: a host that errored but had no
    # state files still gets one marker row.
    for hname in sorted(unreachable - listed_hosts):
        rows.append(
            {
                "name": "-",
                "host": hname,
                "port": "-",
                "image": "-",
                "status": "unreachable",
                "uptime": "-",
                "stale": False,
            }  # fmt: skip
        )
    return rows


def do_list(as_json: bool, local_only: bool = False) -> None:
    migrate_flat_state()
    rt = detect_runtime()
    rows = gather_rows(rt, local_only)
    if as_json:
        print(json.dumps(rows, indent=2))
        return
    table = Table(show_header=True, header_style="bold", box=None, pad_edge=False)
    for col in ("NAME", "HOST", "PORT", "IMAGE", "STATUS", "UPTIME"):
        table.add_column(col)
    for row in rows:
        style = "dim" if row["stale"] else None
        table.add_row(
            str(row["name"]),
            str(row["host"]),
            str(row["port"]),
            str(row["image"]),
            str(row["status"]),
            str(row["uptime"]),
            style=style,
        )
    console.print(table)


def ssh_argv(user: str, host: str, port: str | int, window: str | None = None) -> list[str]:
    # The one true attach command; argv list, never a shell string. Without a
    # window the argv is unchanged (agent-container parity). With one, the remote
    # command is a SINGLE compound string that selects the window (tolerating
    # its absence) THEN attaches — select-window works on a detached session.
    base = ["ssh", f"{user}@{host}", "-p", str(port), "-t"]
    if window:
        return base + [f"tmux select-window -t main:{window} 2>/dev/null; exec tmux attach -t main"]
    return base + ["tmux", "attach", "-t", "main"]


def tmux_nest_warning() -> str | None:
    if os.environ.get("TMUX"):
        return "already inside tmux; detach the INNER session with Ctrl-B Ctrl-B d"
    return None


def resolve_attach_target(
    name: str,
    mode: str,  # "auto" | "local" | "remote"
    user_override: str | None = None,
    host_override: str | None = None,
) -> tuple[str, str, str, str]:
    """Returns (user, host, port, kind). kind is 'local' or 'remote'."""
    # Case-insensitive like agent-container: 'My-Box' resolves the same
    # MY_BOX_* keys (and the same lowercase state file agent-container writes).
    name = validate_name(name.lower())
    user = resolve_ssh_user(user_override)
    if host_override is not None and not SSH_HOST_RE.fullmatch(host_override):
        die(f"invalid ssh host '{host_override}'")
    key = name_to_key(name)

    if mode == "remote" and not HOSTS_CONF.is_file():
        die(f"no hosts config at {HOSTS_CONF} — create it (see docs/agent-container-hosts.example)")

    entry = hosts_entry(name)
    if mode == "remote" or (mode == "auto" and entry is not None):
        if entry is None:
            die(f"no host configured for {name} — add {key}_HOST and {key}_PORT to {HOSTS_CONF}")
        host, port = entry
        return user, host_override or host, port, "remote"

    # Local target: per-host state under the implicit 'local' host. (Attaching to
    # a specific registered host by name is a follow-up; today `up`/`down` carry
    # --host and local attach covers the default-host loop.)
    port = read_state_port(DEFAULT_HOST, name)
    if port is None:
        sf = state_file_for(DEFAULT_HOST, name)
        if mode == "local":
            die(
                f"no local state for {name} at {sf} — is the container "
                f"running? (try: agent-container up {name})"
            )
        die(
            f"no attach target for {name}; checked {HOSTS_CONF} (no {key}_HOST/{key}_PORT) and {sf}"
        )
    host = host_override or os.environ.get("AGENT_CONTAINER_HOST") or "localhost"
    return user, host, port, "local"


def cli_attach(
    name: str,
    mode: str,
    user_override: str | None,
    host_override: str | None,
    window: str | None = None,
) -> None:
    # Validate the window BEFORE resolving/building the command (it is embedded
    # in the remote shell string).
    if window:
        validate_window(window)
    migrate_flat_state()
    user, host, port, _ = resolve_attach_target(name, mode, user_override, host_override)
    if w := tmux_nest_warning():
        warn(w)  # CLI mode: warn and proceed
    session = '"main"' + (f', window "{window}"' if window else "")
    eprint(f"agent-container: {user}@{host}:{port} (tmux session {session})")
    sys.stdout.flush()
    sys.stderr.flush()
    # Full handover: the process is replaced; signals, SIGWINCH, TTY ownership
    # and the exit code belong to ssh — identical to bash `exec ssh`.
    os.execvp("ssh", ssh_argv(user, host, port, window))


def _child_default_sigint() -> None:
    # SIG_IGN survives exec; without this reset a child spawned while the
    # parent shields itself from Ctrl-C would ignore SIGINT too.
    signal.signal(signal.SIGINT, signal.SIG_DFL)


def wizard_handover(
    user: str,
    host: str,
    port: str,
    logs_hint: str = "agent-container logs <name>",
    window: str | None = None,
) -> None:
    """Attach without killing the wizard: inherited-stdio subprocess + cleanup."""
    target = f"{user}@{host}:{port}"
    session = '"main"' + (f', window "{window}"' if window else "")
    eprint(f"agent-container: {target} (tmux session {session})")
    sys.stdout.flush()
    sys.stderr.flush()
    old_int = signal.signal(signal.SIGINT, signal.SIG_IGN)  # Ctrl-C goes to ssh only
    try:
        rc = subprocess.run(
            ssh_argv(user, host, port, window), preexec_fn=_child_default_sigint
        ).returncode
    finally:
        signal.signal(signal.SIGINT, old_int)
        if _ORIG_TERMIOS is not None:
            try:
                termios.tcsetattr(sys.stdin.fileno(), termios.TCSADRAIN, _ORIG_TERMIOS)
            except Exception:
                pass
    if rc == 0:
        console.print(
            f"Detached — tmux session 'main' is still running on {target}; agents keep working."
        )
    elif rc == 255:
        warn(f"ssh could not connect to {target}; check hosts.conf or the container state.")
    else:
        warn(
            f"remote command failed (exit {rc}); likely 'tmux attach' found no session "
            f"'main'. Check the entrypoint via: {logs_hint}"
        )


def do_logs(name: str, follow: bool) -> int:
    validate_name(name)
    rt = detect_runtime()
    argv = [rt, "logs"] + (["-f"] if follow else []) + [container_name(name)]
    try:
        return subprocess.run(argv).returncode  # inherited stdio
    except KeyboardInterrupt:
        return 130


# --- wizard ------------------------------------------------------------------
# Lifecycle pickers (start/stop/purge/logs) list LOCAL containers only;
# remote (hosts.conf) targets appear exclusively in the Attach picker.


def _validate_wiz_name(text: str):  # questionary validator: True or error string
    return True if NAME_RE.fullmatch(text) else "must match [a-z0-9][a-z0-9_-]*"


def _short(cname: str) -> str:
    return cname[len(CONTAINER_PREFIX) :]


def wiz_build(rt: str) -> None:
    tag = questionary.text("image tag", default=IMAGE_NAME).ask()
    if not tag:
        return
    do_build(tag)
    hint("agent-container build" + (f" {tag}" if tag != IMAGE_NAME else ""))


def wiz_up(rt: str) -> None:
    name = questionary.text("container name", validate=_validate_wiz_name).ask()
    if not name:
        return
    migrate_flat_state()
    cname = container_name(name)
    host_rec = implicit_local_host(rt)  # the wizard deploys to the local host

    if cname in host_container_names(host_rec):
        port = read_state_port(DEFAULT_HOST, name)
        log(f"container {cname} already running" + (f" on port {port}" if port else ""))
        return
    if cname in host_container_names(host_rec, include_stopped=True):
        log(f"container {cname} exists but is stopped — remove it first (Stop / remove)")
        return

    env_file = resolve_env_file(name)
    if env_file is None:
        eprint("[agent-container] no env file found; looked in:")
        for c in env_file_candidates(name, Path.cwd()):
            eprint(f"  {c}")
        target = CONFIG_DIR / f"{name}.env"
        if questionary.confirm(f"create/edit {target} in $EDITOR now?", default=True).ask():
            CONFIG_DIR.mkdir(parents=True, exist_ok=True)
            # $EDITOR gets the path only; agent-container never reads env contents.
            subprocess.run([os.environ.get("EDITOR") or "vi", str(target)])
            env_file = resolve_env_file(name)
        if env_file is None:
            log("still no env file; not starting")
            return

    # Optional single host-dir bind (empty = none). Resolved now so a bad path
    # surfaces before the confirmation card; Fatal returns to the menu.
    mounts: list[str] = []
    mount_dir = questionary.text(
        "host directory to mount read-write (empty = none)", default=""
    ).ask()
    if mount_dir:
        mounts = [resolve_bind_mount(mount_dir)]

    card = Table(show_header=False, box=None, pad_edge=False)
    card.add_column(style="bold")
    card.add_column()
    card.add_row("env-file", str(env_file))  # path only, never contents
    card.add_row("port", str(port_for_name(name)))
    card.add_row("volumes", ", ".join(per_container_volumes(name)))
    if mounts:
        card.add_row("bind", mounts[0])
    card.add_row("image", "built on host via compose")
    console.print(card)

    if not questionary.confirm(f"start {cname}?", default=True).ask():
        return
    # Compose build + up (port pre-check + actionable abort inside).
    with deployment_lock(DEFAULT_HOST, name):
        compose_up_exec(DEFAULT_HOST, host_rec, name, env_file, mounts, None, [])
    hint(f"agent-container up {name}" + (f" --mount {mount_dir}" if mounts else ""))


def wiz_attach(rt: str) -> None:
    choices: list[questionary.Choice] = []
    for cname, _image, _status, _uptime in ps_agent_container(rt):
        short = _short(cname)
        port = read_state_port(DEFAULT_HOST, short) or "?"
        choices.append(
            questionary.Choice(f"{short}  [local :{port}]", value=("local", short, "", ""))
        )
    conf = load_hosts_conf()
    for key in sorted(conf):
        if not key.endswith("_HOST"):
            continue
        base = key[: -len("_HOST")]
        host, port = conf[key], conf.get(f"{base}_PORT")
        if not (host and port):
            continue
        rname = base.lower()  # round-trips through name_to_key for the hint
        choices.append(
            questionary.Choice(
                f"{rname}  [remote {host}:{port}]", value=("remote", rname, host, port)
            )
        )
    if not choices:
        log("nothing to attach to: no running local containers and no hosts.conf entries")
        return

    sel = questionary.select("attach to", choices=choices).ask()
    if sel is None:
        return
    if w := tmux_nest_warning():
        warn(w)
        if not questionary.confirm("continue attaching?", default=True).ask():
            return

    # Optional window to select before attaching (empty = session default).
    # Validated now so a bad name aborts back to the menu (Fatal is caught there).
    window = questionary.text("window to select (empty = default)", default="").ask() or None
    if window:
        validate_window(window)

    kind, name, host, port = sel
    user = resolve_ssh_user()
    if kind == "local":
        port = read_state_port(DEFAULT_HOST, name)
        if port is None:  # TOCTOU: state file vanished since the picker rendered
            warn(f"state file for {name} disappeared; is the container still up?")
            return
        host = os.environ.get("AGENT_CONTAINER_HOST") or "localhost"
    wizard_handover(user, host, port, logs_hint=f"agent-container logs {name}", window=window)
    hint(f"agent-container attach {name} --{kind}" + (f" --window {window}" if window else ""))


def wiz_logs(rt: str) -> None:
    rows = ps_agent_container(rt)
    if not rows:
        log("no running agent-container containers")
        return
    sel = questionary.select("logs for", choices=[_short(r[0]) for r in rows]).ask()
    if sel is None:
        return
    log("streaming logs — Ctrl-C returns to the menu")
    old_int = signal.signal(signal.SIGINT, signal.SIG_IGN)  # Ctrl-C stops the tail only
    try:
        subprocess.run([rt, "logs", "-f", container_name(sel)], preexec_fn=_child_default_sigint)
    finally:
        signal.signal(signal.SIGINT, old_int)
    hint(f"agent-container logs {sel}")


def wiz_down(rt: str) -> None:
    rows = ps_agent_container(rt, include_stopped=True)
    if not rows:
        log("no agent-container containers (running or stopped)")
        return
    choices = []
    for cname, _image, status, _uptime in rows:
        short = _short(cname)
        label = f"{short}  [exited]" if status.lower().startswith("exited") else short
        choices.append(questionary.Choice(label, value=short))
    sel = questionary.select("stop / remove", choices=choices).ask()
    if sel is None:
        return
    cname = container_name(sel)

    if not container_exists(rt, cname):  # TOCTOU re-verify
        warn(f"container {cname} no longer exists")
        return
    if quadlet_active(sel):
        warn(
            f"systemd user unit {cname}.service is active; systemd will restart "
            f"the container after removal. Manage it via Quadlet instead."
        )
        if not questionary.confirm("proceed anyway?", default=False).ask():
            return
    if not questionary.confirm(f"stop and remove {cname}?", default=False).ask():
        log("aborted")
        return

    purge = False
    vols = per_container_volumes(sel)
    if questionary.confirm(
        f"also delete ALL per-container volumes ({', '.join(vols)})?", default=False
    ).ask():
        typed = questionary.text(
            f"type the container name ({cname}) to confirm volume deletion"
        ).ask()
        if typed == cname:
            purge = True
        else:
            log("name mismatch — volumes will be preserved")
    with deployment_lock(DEFAULT_HOST, sel):  # serialize vs concurrent lifecycle ops (FR-017)
        down_container(DEFAULT_HOST, implicit_local_host(rt), sel, purge)  # reports + clears state
    hint(f"agent-container down {sel}" + (" --purge" if purge else ""))


def _volume_names(rt: str) -> list[str]:
    out = query([rt, "volume", "ls", "--format", "{{.Name}}"]).stdout.splitlines()
    return [v for v in out if re.fullmatch(re.escape(CONTAINER_PREFIX) + r".+-workspace", v)]


def wiz_purge_volume(rt: str) -> None:
    containers = {row[0] for row in ps_agent_container(rt, include_stopped=True)}
    orphans = []
    for vol in _volume_names(rt):
        short = vol[len(CONTAINER_PREFIX) : -len("-workspace")]
        if container_name(short) not in containers:
            orphans.append(vol)
    if not orphans:
        log("no orphaned agent-container volumes")
        return
    sel = questionary.select("purge orphaned volume", choices=orphans).ask()
    if sel is None:
        return
    typed = questionary.text(f"type the volume name ({sel}) to confirm deletion").ask()
    if typed != sel:
        log("name mismatch — nothing removed")
        return
    if sel not in _volume_names(rt):  # TOCTOU re-verify
        warn(f"volume {sel} no longer exists")
        return
    if query([rt, "volume", "rm", sel]).returncode != 0:
        warn(f"volume {sel} not removed (in use)")
    else:
        log(f"volume {sel} purged")
    hint(f"{rt} volume rm {sel}")


def wiz_keys(rt: str) -> None:
    rows = ps_agent_container(rt, include_stopped=False)  # running only
    if not rows:
        log("no running agent-container containers")
        return
    sel = questionary.select("inject SSH keys into", choices=[_short(r[0]) for r in rows]).ask()
    if sel is None:
        return
    hk = questionary.text("host key path (unencrypted ed25519 private key; blank to skip)").ask()
    ak = questionary.text("authorized public key path (blank to skip)").ask()
    host_key = Path(hk).expanduser() if hk else None
    authorized = [Path(ak).expanduser()] if ak else []
    if host_key is None and not authorized:
        log("nothing to inject")
        return
    inject_keys(rt, sel, host_key, authorized)
    hint(
        f"agent-container keys {sel}"
        + (f" --host-key {hk}" if hk else "")
        + (f" --authorized-key {ak}" if ak else "")
    )


_MENU: list[tuple[str, Callable[[str], None] | None]] = [
    ("Provision (build image)", wiz_build),
    ("Start a container", wiz_up),
    ("Attach", wiz_attach),
    ("Inject SSH keys", wiz_keys),
    ("Logs", wiz_logs),
    ("Stop / remove", wiz_down),
    ("Purge orphaned volume", wiz_purge_volume),
    ("Quit", None),
]


def wizard_loop() -> int:
    if not is_tty():
        eprint(
            "[agent-container] no TTY; interactive wizard unavailable. Use subcommands (agent-container --help)."
        )
        return 2
    try:
        rt = detect_runtime()
    except Fatal as e:
        eprint(f"[agent-container] FATAL: {e}")
        return 1
    actions = dict(_MENU)
    while True:
        try:
            running = len(ps_agent_container(rt))
            hosts = "present" if HOSTS_CONF.is_file() else "absent"
            console.print(
                f"runtime: {rt} | running local containers: {running} | hosts.conf: {hosts}",
                style="dim",
            )
            choice = questionary.select(
                "agent-container — what now?", choices=[t for t, _ in _MENU]
            ).ask()
            if choice is None or choice == "Quit":
                return 0
            try:
                action = actions[choice]
                if action is not None:
                    action(rt)
            except Fatal as e:
                eprint(f"[agent-container] ERROR: {e}")  # back to the menu, not exit
        except KeyboardInterrupt, EOFError:
            return 0


# --- entry -------------------------------------------------------------------

app = typer.Typer(
    add_completion=False,
    no_args_is_help=False,
    pretty_exceptions_enable=False,
    help="Interactive wizard + CLI for agent-container containers.",
)


def fatal_exit(e: Fatal) -> NoReturn:
    eprint(f"[agent-container] FATAL: {e}")
    raise typer.Exit(1)


def run_self_test() -> int:
    import doctest

    failures, tests = doctest.testmod(
        sys.modules[__name__], optionflags=doctest.IGNORE_EXCEPTION_DETAIL
    )
    # Hard-coded port corpus: pins the deterministic port hash against regressions.
    corpus = {
        "acme": 2206,
        "blog": 2220,
        "scratch": 2244,
        "my-box": 2204,
        "a": 2297,
        "devbox123": 2298,
    }
    bad = {n: (port_for_name(n), want) for n, want in corpus.items() if port_for_name(n) != want}
    keys_ok = name_to_key("my-box") == "MY_BOX" and name_to_key("acme") == "ACME"
    ok = failures == 0 and not bad and keys_ok
    print(f"doctests: {tests - failures}/{tests} passed")
    print(f"port corpus: {'ok' if not bad else f'MISMATCH {bad}'}")
    print(f"key derivation: {'ok' if keys_ok else 'MISMATCH'}")
    print("self-test:", "PASS" if ok else "FAIL")
    return 0 if ok else 1


@app.command()
def build(
    tag: str = typer.Argument(IMAGE_NAME, help="Image tag to build."),
    context: Path | None = typer.Option(
        None,
        "--context",
        help="Docker build context (repo checkout). Defaults to AGENT_CONTAINER_REPO "
        "or the auto-detected checkout; required when installed from PyPI.",
    ),
) -> None:
    """Build the image from a repo checkout (streams build output)."""
    do_build(tag, context)


@app.command()
def up(
    name: str = typer.Argument(..., help="Container short name (agent-container-<name>)."),
    host: str | None = typer.Option(
        None, "--host", help="Deploy to this registered host (default: the registry default)."
    ),
    env_file: Path | None = typer.Option(
        None, "--env-file", help="Bypass env-file resolution; path must exist."
    ),
    mount: list[str] = typer.Option(
        None,
        "--mount",
        help="Bind-mount a host dir read-write (repeatable): HOSTDIR[:CONTAINERPATH]. "
        "Default CONTAINERPATH is /workspace/<basename>. Lima: host dir must be "
        "under a writable Lima mount.",
    ),
    host_key: Path | None = typer.Option(
        None,
        "--host-key",
        help="Inject an OpenSSH ed25519 PRIVATE host key so the container adopts "
        "a known SSH identity (persisted on the ~/.ssh volume). Must be "
        "unencrypted.",
    ),
    authorized_key: list[Path] = typer.Option(
        None,
        "--authorized-key",
        help="Inject an SSH public key into authorized_keys (repeatable), so you "
        "can attach without a manual exec. Persisted on the ~/.ssh volume.",
    ),
) -> None:
    """Start container agent-container-NAME (detached)."""
    if env_file is not None and not env_file.is_file():
        raise typer.BadParameter(f"--env-file {env_file} does not exist")
    do_up(name, host, env_file, mount or [], host_key, authorized_key or [])


@app.command()
def keys(
    name: str = typer.Argument(..., help="Container short name."),
    host_key: Path | None = typer.Option(
        None,
        "--host-key",
        help="Install an unencrypted ed25519 PRIVATE host key into the running "
        "container and reload sshd (persists on the ~/.ssh volume).",
    ),
    authorized_key: list[Path] = typer.Option(
        None,
        "--authorized-key",
        help="Append an SSH public key to authorized_keys in the running "
        "container (repeatable; deduped; persists on the ~/.ssh volume).",
    ),
) -> None:
    """Inject SSH host key / authorized keys into a RUNNING container (no recreate)."""
    cli_keys(name, host_key, authorized_key or [])


@app.command()
def down(
    name: str = typer.Argument(..., help="Container short name."),
    host: str | None = typer.Option(
        None, "--host", help="Host the container runs on (default: the registry default)."
    ),
    purge: bool = typer.Option(
        False,
        "--purge",
        help="Also delete all seven per-container volumes (workspace + agent logins, shell env, tmux config, ssh).",
    ),
    yes: bool = typer.Option(False, "-y", "--yes", help="Skip confirmation."),
) -> None:
    """Stop and remove container agent-container-NAME (dispose; volumes kept unless --purge)."""
    cli_down(name, purge, yes, host)


@app.command()
def stop(
    name: str = typer.Argument(..., help="Container short name."),
    host: str | None = typer.Option(None, "--host", help="Host the container runs on."),
) -> None:
    """Stop (pause/reclaim) agent-container-NAME — retained with its volumes; `start` resumes it."""
    do_stop(name, host)


@app.command()
def start(
    name: str = typer.Argument(..., help="Container short name."),
    host: str | None = typer.Option(None, "--host", help="Host the container runs on."),
) -> None:
    """Start a stopped agent-container-NAME (no rebuild, no recreate)."""
    do_start(name, host)


@app.command()
def redeploy(
    name: str = typer.Argument(..., help="Container short name."),
    host: str | None = typer.Option(None, "--host", help="Host the container runs on."),
    env_file: Path | None = typer.Option(
        None, "--env-file", help="Bypass env-file resolution; path must exist."
    ),
    mount: list[str] = typer.Option(
        None, "--mount", help="Bind-mount a host dir (repeatable), as for `up`."
    ),
    host_key: Path | None = typer.Option(
        None, "--host-key", help="Inject an ed25519 PRIVATE host key (as for `up`)."
    ),
    authorized_key: list[Path] = typer.Option(
        None, "--authorized-key", help="Inject an SSH public key (repeatable)."
    ),
) -> None:
    """Rebuild the image and recreate agent-container-NAME, preserving its volumes.
    Deliberately non-idempotent: always rebuilds + recreates, even with no change."""
    if env_file is not None and not env_file.is_file():
        raise typer.BadParameter(f"--env-file {env_file} does not exist")
    do_redeploy(name, host, env_file, mount or [], host_key, authorized_key or [])


@app.command()
def wipe(
    name: str = typer.Argument(..., help="Container short name."),
    host: str | None = typer.Option(None, "--host", help="Host the container runs on."),
    yes: bool = typer.Option(False, "-y", "--yes", help="Skip confirmation."),
) -> None:
    """WIPE agent-container-NAME: remove the container, its volumes, AND its built image (confirmed)."""
    do_wipe(name, yes, host)


@app.command()
def purge(
    name: str = typer.Argument(..., help="Container short name."),
    yes: bool = typer.Option(False, "-y", "--yes", help="Skip confirmation."),
) -> None:
    """Sugar for: down NAME --purge."""
    cli_down(name, purge=True, yes=yes)


@app.command(name="list")
def list_cmd(
    as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),
    local_only: bool = typer.Option(
        False,
        "--local",
        help="Skip remote round-trips: list only the local runtime + per-host state files (fast).",
    ),
) -> None:
    """List agent-container containers, reconciled live against each registered
    host (a --local view skips remote queries)."""
    do_list(as_json, local_only)


# The `host` group manages WHERE containers run (registered targets). Container
# lifecycle (up/down/attach/logs) stays as bare top-level verbs; a container is
# addressed by name and lives on a host chosen with --host (default: the registry
# default). There is deliberately no `host up`/`host down` — those are container
# operations, not host operations. Deprovisioning a host is `host rm --destroy`
# (guarded: refused while containers remain, or for hosts the tool did not create).
host_app = typer.Typer(
    no_args_is_help=True,
    help="Manage deployment hosts (the machines/contexts where containers run).",
)
app.add_typer(host_app, name="host")


@host_app.command("add")
def host_add(
    name: str = typer.Argument(..., help="Host short name (registry key)."),
    driver: str = typer.Option("docker", "--driver", help="Runtime driver: docker or podman."),
    docker_context: str | None = typer.Option(
        None,
        "--docker-context",
        help="Existing docker context (local endpoint or ssh://user@host).",
    ),
    connection: str | None = typer.Option(
        None, "--connection", help="Podman system connection name (for --driver podman)."
    ),
    address: str | None = typer.Option(
        None, "--address", help="Attach address override (default: derived from the context)."
    ),
    make_default: bool = typer.Option(
        False, "--default", help="Make this the default deploy target."
    ),
    provider: str | None = typer.Option(
        None, "--provider", help="Cloud provider to provision a server on (e.g. hetzner)."
    ),
    create: bool = typer.Option(
        False, "--create", help="Allocate a NEW cloud server (billable). Requires --provider."
    ),
    reuse: bool = typer.Option(
        False,
        "--reuse",
        help="Register an EXISTING cloud server (no allocation); use with --docker-context ssh://…",
    ),
    server_type: str | None = typer.Option(
        None, "--server-type", help="Cloud server type (e.g. cax11). Provider-specific."
    ),
    location: str | None = typer.Option(
        None, "--location", help="Cloud location (e.g. nbg1). Provider-specific."
    ),
    ssh_key: str | None = typer.Option(
        None, "--ssh-key", help="Existing provider SSH key id/name to also attach (optional)."
    ),
    ssh_pubkey: Path | None = typer.Option(
        None,
        "--ssh-pubkey",
        help="Operator SSH PUBLIC key to authorize on the new server "
        "(default: ~/.ssh/id_ed25519.pub).",
    ),
) -> None:
    """Register a container-runtime host (where `up --host NAME` deploys), or
    provision a cloud server with --provider hetzner --create."""
    cli_host_add(
        name,
        driver,
        docker_context or connection,
        address,
        make_default,
        provider=provider,
        create=create,
        reuse=reuse,
        server_type=server_type,
        location=location,
        ssh_key=ssh_key,
        ssh_pubkey=ssh_pubkey,
    )


@host_app.command("ls")
def host_ls(
    as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),
) -> None:
    """List registered hosts (name, driver, context, address, default)."""
    do_host_ls(as_json)


@host_app.command("show")
def host_show(
    name: str = typer.Argument(..., help="Host short name."),
    as_json: bool = typer.Option(False, "--json", help="Emit machine-readable JSON."),
) -> None:
    """Show one host's full record (driver, context, address, provisioning state)."""
    do_host_show(name, as_json)


@host_app.command("rm")
def host_rm(
    name: str = typer.Argument(..., help="Host short name to remove."),
    destroy: bool = typer.Option(
        False,
        "--destroy",
        help="Also deprovision the cloud server (tool-created hosts only; irreversible).",
    ),
    yes: bool = typer.Option(False, "-y", "--yes", help="Skip the confirmation prompt."),
) -> None:
    """Remove a host from the registry. With --destroy also deallocate its cloud
    server — refused if the server still hosts containers or the tool didn't create it."""
    cli_host_rm(name, destroy, yes)


@app.command()
def attach(
    name: str = typer.Argument(..., help="Container short name."),
    local: bool = typer.Option(False, "--local", help="Force local target (state file)."),
    remote: bool = typer.Option(False, "--remote", help="Force remote target (hosts.conf)."),
    user: str | None = typer.Option(
        None, "--user", help="SSH user (default: AGENT_CONTAINER_USER or dev)."
    ),
    host: str | None = typer.Option(None, "--host", help="Override the resolved host."),
    window: str | None = typer.Option(
        None, "--window", "-w", help="Select tmux window NAME (session 'main') before attaching."
    ),
) -> None:
    """ssh + tmux attach to container NAME (local state file or hosts.conf)."""
    if local and remote:
        raise typer.BadParameter("--local and --remote are mutually exclusive")
    mode = "local" if local else ("remote" if remote else "auto")
    cli_attach(name, mode, user, host, window)


@app.command()
def logs(
    name: str = typer.Argument(..., help="Container short name."),
    no_follow: bool = typer.Option(False, "--no-follow", help="Print logs without following."),
) -> None:
    """Tail container logs."""
    raise typer.Exit(do_logs(name, follow=not no_follow))


@app.command()
def menu() -> None:
    """Interactive wizard (same as bare agent-container)."""
    raise typer.Exit(wizard_loop())


def _completion_script(shell: str) -> str:
    """Return the bash/zsh completion script text.

    Prefers the on-disk checkout (dev / `uv run --script`, where REPO_ROOT
    resolves via the __file__ marker); falls back to the completions bundled as
    package data in a non-editable PyPI install (where REPO_ROOT is None).
    """
    if REPO_ROOT is not None:
        p = REPO_ROOT / "completions" / f"agent-container.{shell}"
        if p.is_file():
            return p.read_text()
    try:
        import importlib.resources as ir

        res = ir.files("agent_container").joinpath(f"completions/agent-container.{shell}")
        if res.is_file():
            return res.read_text()
    except ModuleNotFoundError, ImportError, FileNotFoundError, TypeError, AttributeError:
        pass
    die(f"completion script for '{shell}' not found (need a checkout or an installed package)")


@app.command()
def completions(
    # Optional + manual validation so a bad/missing shell routes through die()
    # (exit 1, '[agent-container] FATAL:' style), not Typer's exit-2 required-arg error.
    shell: str = typer.Argument("", help="Shell to emit completion for: bash or zsh."),
) -> None:
    """Print the completion script for bash or zsh (from a checkout or package data)."""
    if shell not in ("bash", "zsh"):
        die("usage: agent-container completions <bash|zsh>")
    sys.stdout.write(_completion_script(shell))


def _resolve_version() -> str:
    """The tool's version, single-sourced from pyproject.toml. When installed
    (wheel), it is read from package metadata; when run via `uv run --script`
    from a checkout, it is read from pyproject directly — no hardcoded constant
    to drift from the release version release-please bumps."""
    try:
        from importlib.metadata import PackageNotFoundError, version

        return version("agent-container")
    except PackageNotFoundError:
        pass
    if REPO_ROOT is not None:
        try:
            import tomllib

            with (REPO_ROOT / "pyproject.toml").open("rb") as f:
                return tomllib.load(f)["project"]["version"]
        except Exception:
            pass
    return "0.0.0+unknown"


def _version_callback(value: bool) -> None:
    if value:
        typer.echo(_resolve_version())
        raise typer.Exit()


@app.callback(invoke_without_command=True)
def main(
    ctx: typer.Context,
    self_test: bool = typer.Option(
        False, "--self-test", help="Run doctests + port-hash corpus checks."
    ),
    version: bool = typer.Option(
        False,
        "--version",
        callback=_version_callback,
        is_eager=True,
        help="Show the agent-container version and exit.",
    ),
) -> None:
    if self_test:
        raise typer.Exit(run_self_test())
    if ctx.invoked_subcommand is None:
        raise typer.Exit(wizard_loop())


def cli() -> None:
    """Console-script entry point for `uv tool install` (see pyproject.toml).

    uv's generated launcher does `from agent_container import cli; sys.exit(cli())`,
    so the Fatal -> exit-1 handling that used to live only in the __main__ guard
    is hoisted here. NOT named `main` — that is already the Typer @app.callback.
    """
    try:
        app()
    except Fatal as e:
        eprint(f"[agent-container] FATAL: {e}")
        sys.exit(1)


if __name__ == "__main__":
    cli()
