//! `hivectl` — operator-facing host CLI for hyperhive.
//!
//! Sibling binary to the `hive-c0re` daemon. Where `hive-c0re`'s
//! subcommands focus on the broker / approval / topology surface
//! (`spawn`, `kill`, `rebuild`, `approve` …), `hivectl` covers
//! host-side administration — both manual provisioning operations
//! that work without the daemon running (forge, matrix, gateway) AND
//! daemon-assisted agent management (agents restart/restart-all) that
//! goes through the host admin socket.
//!
//! Verbs read configuration off the same on-disk paths c0re uses
//! (`/var/lib/hyperhive/forge-core-token`,
//! `/var/lib/hyperhive/matrix-register-token`, per-agent state
//! dirs) and reuse the `forge` / `matrix` modules from the
//! `hive-c0re` lib — single source of truth, no duplication.

#[cfg(unix)]
use std::os::unix::process::CommandExt as _;
use std::path::{Path, PathBuf};

use anyhow::{Context as _, Result, bail};
use clap::{Args, Parser, Subcommand};
use hive_c0re::coordinator::Coordinator;

/// Rebuild-queue DAG progress rendering (`wait_for_dags` + the spinner /
/// plain renderers), split out to keep this file manageable. `#[path]` keeps
/// the file under `bin/hivectl/` (a subdir cargo won't treat as its own
/// binary) rather than the sibling `bin/dag_progress.rs` a bare `mod` maps to.
#[path = "hivectl/dag_progress.rs"]
mod dag_progress;
use dag_progress::wait_for_dags;

#[derive(Parser)]
#[command(
    name = "hivectl",
    about = "hyperhive host CLI — operator-facing administration",
    long_about = "\
Sibling to the `hive-c0re` daemon binary. Covers host-side admin \
operations that don't go through the broker — manual user \
provisioning on the bundled forge + matrix containers, plus future \
recovery / debugging verbs.\
"
)]
struct Cli {
    /// Path to the hive-c0re host admin socket, used by the daemon-assisted
    /// verbs (`agents`, `stop`, `start`). Global: accepted before or after
    /// the subcommand. Verbs that don't talk to the daemon ignore it.
    #[arg(long, global = true, default_value = DEFAULT_HOST_SOCKET)]
    socket: PathBuf,
    #[command(subcommand)]
    cmd: Cmd,
}

#[derive(Subcommand)]
enum Cmd {
    /// Forgejo user provisioning. Manual entry point to the same
    /// idempotent flow c0re runs automatically at boot
    /// (`forge::ensure_all`) — useful for recovery, ad-hoc reprovisioning,
    /// or single-agent fixes without bouncing the daemon.
    Forge {
        #[command(subcommand)]
        cmd: ForgeCmd,
    },
    /// matrix-tuwunel user provisioning. Manual entry point to the same
    /// idempotent flow c0re runs automatically at boot
    /// (`matrix::ensure_all`) — useful when the boot-time sweep skipped
    /// an agent (e.g. matrix container wasn't up yet) or to re-register
    /// after wiping a token file.
    Matrix {
        #[command(subcommand)]
        cmd: MatrixCmd,
    },
    /// GitHub account provisioning: write an operator-supplied personal
    /// access token (PAT) into an agent's `github-token` state file so its
    /// `gh` wrapper + git credential helper can authenticate. Unlike
    /// forge/matrix there is no account creation — the operator supplies a
    /// PAT for an existing GitHub account. A CLI alternative to the dashboard
    /// credentials tab; the integration is on by default
    /// (`hyperhive.github.enable`), so no per-agent config is needed.
    Github {
        #[command(subcommand)]
        cmd: GithubCmd,
    },
    /// Gateway htpasswd user management. Add, remove, or list users in
    /// an htpasswd file used by the gateway's HTTP Basic auth
    /// (`services.hyperhive.gateway.auth`). Credentials are stored as
    /// `BCrypt` hashes — no extra service or PAM required.
    Gateway {
        #[command(subcommand)]
        cmd: GatewayCmd,
    },
    /// Agent container management. Requires the hive-c0re daemon to be
    /// running (connects to the host admin socket).
    Agents {
        #[command(subcommand)]
        cmd: AgentsCmd,
    },
    /// WireGuard inter-hive mesh setup helpers (`services.hyperhive.swarm`).
    ///
    /// One-time-setup convenience so nobody has to remember the `wg` dance:
    /// `wg init` generates + stores this hive's private key and prints the
    /// public key plus the nix snippet to enable the mesh; `wg peer` prints
    /// the snippet to add a remote hive; `wg status` wraps `wg show`. The
    /// verbs own the imperative state (the key file); the printed nix goes
    /// into the operator's host config (kept in git), so nothing here mutates
    /// declarative config behind the operator's back.
    Wg {
        #[command(subcommand)]
        cmd: WgCmd,
    },
    /// Generate the federation peer-config block for THIS hive — the nix
    /// a peer operator pastes into their `services.hyperhive.swarm.peers`
    /// to trust + reach this hive. Emits `caCert` (+ a `cp` line for the
    /// cert) when this hive serves a self-signed CA, the WireGuard public
    /// key when the mesh key exists, and the `wireguard{Address,Endpoint}`
    /// you pass. The hive's own domain is filled in automatically from the
    /// running daemon (`services.hyperhive.domain`). Reads local state (the
    /// TLS CA cert, the wg key); never mutates. `wg init` calls this at the
    /// end, so a fresh mesh setup prints the hand-over block too.
    PeerConfig {
        /// This hive's WireGuard mesh address (e.g. `10.42.0.1/32`),
        /// emitted as `wireguardAddress`. Omit when not running the mesh.
        #[arg(long)]
        wg_address: Option<String>,
        /// This hive's public WireGuard endpoint (`host:port`), emitted as
        /// `wireguardEndpoint`. Omit when peers dial in / no mesh.
        #[arg(long)]
        wg_endpoint: Option<String>,
    },
    /// Open an interactive Claude session inside an agent container.
    ///
    /// Runs claude as the agent user from its state dir with the harness's
    /// settings / MCP / system prompt. Bare `choom <name>` is a fresh
    /// session; `--resume <session-id>` rejoins a prior one. Never
    /// collides with the harness's live session. Requires root + a running
    /// container. See `docs/tools/hivectl.md` (Choom) for details.
    Choom {
        /// Agent name (e.g. `damocles`, `iris`).
        name: String,
        /// Resume a prior claude session by its session id, passed
        /// through as `claude --resume <value>` (claude's `--continue`
        /// takes no value — it resumes the cwd's latest session, which
        /// is the harness's, so choom never uses it; this flag matches
        /// the claude flag it maps to). Omit for a fresh blank session.
        /// A value is required when the flag is given.
        #[arg(long = "resume", value_name = "SESSION")]
        resume_session: Option<String>,
    },
    /// Stop containers hive-wide in one operator action. Bare `hivectl
    /// stop` stops **everything** — all sub-agents plus the ci, forge,
    /// gateway, and matrix infra containers. Narrow it with scope flags:
    /// `--agents` (all sub-agents), `--ci` / `--forge` / `--gateway` /
    /// `--matrix` (named infra), and `--agent <name>` (repeatable) for
    /// specific sub-agents. Flags are additive (e.g. `--agents --matrix`).
    /// Requires the hive-c0re daemon (connects to the host admin socket).
    /// hive-c0re itself is never stopped — it services the request.
    Stop {
        #[command(flatten)]
        scope: ScopeArgs,
        /// Gracefully quiesce each agent before stopping, instead of a
        /// hard stop. Each agent gets a graceful-stop DAG on the job
        /// queue: the harness is signalled, runs one stop-checkpoint
        /// turn to flush durable `/state`, drains, then the container
        /// is stopped (bounded by a 3-min timeout that falls back to a
        /// hard stop). All drains overlap. Applies to agents only.
        #[arg(long)]
        graceful: bool,
        /// Return immediately after the stop DAGs are queued instead of
        /// waiting for them with live per-node progress.
        #[arg(long)]
        no_wait: bool,
    },
    /// Start containers hive-wide — the inverse of `hivectl stop`. Bare
    /// `hivectl start` restores the previously-running agents from the
    /// last broad-scope `stop`, or starts everything if no prior stop
    /// exists; the same scope flags as `stop` narrow it (`--agents`,
    /// `--ci`, `--forge`, `--gateway`, `--matrix`, `--agent <name>`). A
    /// targeted `--agent <name>` start is never filtered. Requires the
    /// hive-c0re daemon.
    Start {
        #[command(flatten)]
        scope: ScopeArgs,
        /// Return immediately after the start DAGs are queued instead
        /// of waiting for them with live per-node progress.
        #[arg(long)]
        no_wait: bool,
    },
    /// Restart containers hive-wide — `stop` then `start` over the same
    /// scope. Bare `hivectl restart` restarts **everything** (all sub-agents
    /// plus the ci/forge/gateway/matrix infra containers); the same scope
    /// flags as `stop`/`start` narrow it (`--agents`, `--ci`, `--forge`,
    /// `--gateway`, `--matrix`, `--agent <name>`). If the stop phase reports
    /// a failure the start phase is skipped so the operator can investigate.
    /// Requires the hive-c0re daemon.
    Restart {
        #[command(flatten)]
        scope: ScopeArgs,
        /// Gracefully quiesce each agent on the stop half (see
        /// `stop --graceful`). Applies to agents only.
        #[arg(long)]
        graceful: bool,
    },
    /// Per-agent disk accounting + optional quotas via btrfs qgroups.
    ///
    /// Opt-in: `quota enable` turns on btrfs qgroup accounting for the
    /// agent-state filesystem (a one-time, I/O-heavy rescan — that's why
    /// it isn't automatic). Then `quota show` reports per-agent usage and
    /// `quota limit` caps an agent. No-op on non-btrfs hosts. Operates on
    /// agent state subvolumes (created by the btrfs-subvolume migration);
    /// agents still on a plain dir report no qgroup usage.
    Quota {
        #[command(subcommand)]
        cmd: QuotaCmd,
    },
    /// btrfs subvolume management for agent state dirs.
    ///
    /// New agents get a btrfs subvolume state root automatically (when the
    /// host FS is btrfs); agents that predate that are left on plain dirs.
    /// `subvol upgrade <agent>` opts an existing plain-dir agent into the
    /// subvolume feature set (snapshots, per-subvol usage/quota, migration)
    /// by migrating its state dir in place. Requires the hive-c0re daemon
    /// (for the stop/start) and root (for the privileged migration).
    Subvol {
        #[command(subcommand)]
        cmd: SubvolCmd,
    },
    /// Print (and best-effort open in a browser) a hive web surface URL.
    ///
    /// Resolves the URL from the running daemon (`HostRequest::Urls`), so
    /// custom forge / matrix domains work without guessing `forge.<domain>`.
    /// Prints the URL unconditionally — the reliable core, since the host
    /// is usually headless / driven over SSH where `xdg-open` is a no-op —
    /// then tries `xdg-open` as a convenience. Bare `hivectl open` opens the
    /// operator dashboard.
    Open {
        /// Which surface to open. Defaults to the operator dashboard.
        #[arg(value_enum, default_value_t = OpenTarget::Home)]
        target: OpenTarget,
    },
    /// Emit the full CLI reference as `CommonMark` to stdout.
    ///
    /// Hidden tooling command (not part of day-to-day operator admin):
    /// walks this binary's own clap command tree and renders every verb,
    /// flag, and help string as markdown. Used by the docs build to keep
    /// the published `hivectl` reference in lockstep with the code — no
    /// hand-maintained command list to drift out of date.
    #[command(hide = true)]
    MarkdownDocs,
    /// Generate a shell completion script for `hivectl` and print it to
    /// stdout.
    ///
    /// Pipe it into your shell's completion path — e.g. for zsh:
    /// `hivectl completions zsh > ~/.zsh/completions/_hivectl` (with that
    /// dir on `$fpath`). The hyperhive NixOS module installs the zsh script
    /// system-wide automatically, so this is mainly for ad-hoc / other-shell
    /// use. Supports bash, zsh, fish, elvish, and powershell.
    Completions {
        /// Shell to emit completions for.
        shell: clap_complete::Shell,
    },
}

/// Which hive web surface `hivectl open` targets.
#[derive(Copy, Clone, Debug, clap::ValueEnum)]
enum OpenTarget {
    /// The operator dashboard (`https://<domain>/`).
    Home,
    /// The forge (Forgejo) web UI.
    Forge,
    /// The matrix GUI (fluffychat).
    Matrix,
}

/// Shared scope flags for `hivectl stop` / `hivectl start`. With no flag
/// set the verb targets **everything** (all sub-agents + every controllable
/// infra container). Setting any flag restricts to the selected classes,
/// additively.
// One bool per selectable container class, each mapping 1:1 to a clap flag;
// orthogonal toggles, not a state machine — hence the bools allow (mirrors
// `hive_host_sock::LifecycleScope`).
#[allow(clippy::struct_excessive_bools)]
#[derive(Args)]
struct ScopeArgs {
    /// All sub-agent containers.
    #[arg(long)]
    agents: bool,
    /// A specific sub-agent by name. Repeatable: `--agent a --agent b`.
    #[arg(long = "agent", value_name = "NAME")]
    agent: Vec<String>,
    /// The CI runner container (`hive-ci`).
    #[arg(long)]
    ci: bool,
    /// The forge container (`hive-forge`).
    #[arg(long)]
    forge: bool,
    /// The gateway container (`hive-gateway`).
    #[arg(long)]
    gateway: bool,
    /// The matrix container (`hive-matrix`).
    #[arg(long)]
    matrix: bool,
}

impl ScopeArgs {
    fn to_scope(&self) -> hive_host_sock::LifecycleScope {
        hive_host_sock::LifecycleScope {
            agents: self.agents,
            agent_names: self.agent.clone(),
            ci: self.ci,
            forge: self.forge,
            gateway: self.gateway,
            matrix: self.matrix,
        }
    }
}

#[derive(Subcommand)]
enum ForgeCmd {
    /// Create or refresh the Forgejo account + token for `<name>`.
    ///
    /// When `<name>` matches an existing agent (i.e. it has a state
    /// dir under `/var/lib/hyperhive/agents/`), persists the token to
    /// `<state>/forge-token` (idempotent: re-mints + rewrites every
    /// call so the on-disk scope matches the current
    /// `forge::TOKEN_SCOPES`).
    ///
    /// When `<name>` is **not** an agent (a human or any other
    /// non-container account), creates the forgejo user and prints the
    /// freshly-minted token to stdout — no `/var/lib/hyperhive/agents/`
    /// directory is created for the user.
    ///
    /// Without `--password` / `--password-stdin` a random throwaway is
    /// used (fine for agents — they auth by token via tea / hive-forge).
    /// Set a password to log into the forge web UI afterwards.
    /// `--password` is idempotent: re-running with the same value sets
    /// the same password (covers password resets on already-created
    /// accounts since `forgejo admin user create` silently no-ops once
    /// the user exists).
    CreateUser {
        /// Forgejo username. For agents: the container/agent name
        /// (`<n>` in `h-<n>`; manager uses the literal `manager`).
        /// For humans: any forgejo username — `mara`, `damocles`, etc.
        name: String,
        /// Set the account password to this string instead of a random
        /// throwaway. Use this for operator accounts that need to log
        /// into the forge web UI. Mutually exclusive with
        /// `--password-stdin`. WARNING: the password is visible in
        /// shell history + process listings; prefer `--password-stdin`
        /// for anything sensitive.
        #[arg(long)]
        password: Option<String>,
        /// Read the password from stdin (single line, trailing newline
        /// stripped) instead of an inline flag. Mutually exclusive with
        /// `--password`.
        #[arg(long, conflicts_with = "password")]
        password_stdin: bool,
    },
}

#[derive(Subcommand)]
enum MatrixCmd {
    /// Create or refresh the matrix account + access token for `<name>`.
    ///
    /// When `<name>` matches an existing agent (i.e. it has a state
    /// dir under `/var/lib/hyperhive/agents/`), persists the token to
    /// `<state>/matrix-token`. Skips registration when the file is
    /// already populated; delete it to force re-registration.
    ///
    /// When `<name>` is **not** an agent (a human or any other
    /// non-container account), registers the matrix user and prints
    /// the freshly-minted access token to stdout — no
    /// `/var/lib/hyperhive/agents/` directory is created for the user.
    ///
    /// Without `--password` / `--password-stdin` a random throwaway is
    /// used (fine for agents — they auth by `access_token`, never by
    /// password). Set a password to log into a matrix web client
    /// afterwards.
    CreateUser {
        /// Matrix localpart. For agents: the container/agent name.
        /// For humans: any matrix localpart — `mara`, `damocles`, etc.
        name: String,
        /// Set the account password to this string instead of a random
        /// throwaway. Use this for operator accounts that need to log
        /// into matrix web clients via `m.login.password`. Mutually
        /// exclusive with `--password-stdin`. WARNING: the
        /// password is visible in shell history + process listings;
        /// prefer `--password-stdin` for anything sensitive.
        #[arg(long)]
        password: Option<String>,
        /// Read the password from stdin (single line, trailing newline
        /// stripped) instead of an inline flag. Mutually exclusive with
        /// `--password`.
        #[arg(long, conflicts_with = "password")]
        password_stdin: bool,
    },
    /// Provision (or re-provision) the hive system admin matrix account
    /// (`@hive:<server>`). hive-c0re runs this automatically on startup
    /// before the agent sweep so the account is the first registered
    /// user — Conduit/tuwunel grants admin rights to the first user.
    /// Run manually to recover a missing admin token file.
    SyncAdmin,
    /// Promote a matrix user to homeserver admin via the admin API.
    /// Uses the hive system admin token at
    /// `/var/lib/hyperhive/matrix/admin-token`. The `server_name` is
    /// discovered automatically from the running homeserver.
    PromoteUser {
        /// Matrix localpart of the user to promote (e.g. `argus`).
        name: String,
    },
    /// Reset a matrix user's password via the admin API and persist the
    /// new password to `/var/lib/hyperhive/matrix/creds/<name>-password`
    /// so the next `ensure_user_for` (or `create-user`) can re-login.
    ///
    /// After this command succeeds, run `hivectl matrix create-user
    /// <name>` to mint a fresh access token for the agent.
    ResetPassword {
        /// Matrix localpart of the account to reset (e.g. `argus`).
        name: String,
    },
    /// Invite a matrix user to the hive Space (default) or a specific
    /// room. Uses the hive admin token; the admin account must be a
    /// member of the target room with invite power (it owns the hive
    /// Space). Idempotent — already-member / already-invited is a no-op.
    Invite {
        /// User to invite: a full id (`@mara:server`) or a bare
        /// localpart (qualified with the homeserver's `server_name`).
        user: String,
        /// Target room id (`!abc:server`) or alias (`#name:server`).
        /// Omit to invite to the hive Space.
        #[arg(long)]
        room: Option<String>,
    },
}

#[derive(Subcommand)]
enum GithubCmd {
    /// Write a GitHub PAT into `<agent>`'s state dir (`github-token`, 0600,
    /// agent-owned) via hive-priv. The GitHub integration is on by default
    /// (`hyperhive.github.enable`), so the `gh` wrapper + git credential
    /// helper pick the token up with no per-agent config. The token is read
    /// live at invocation, so no rebuild/restart is needed. Prefer
    /// `--token-stdin`: an inline `--token` is visible in shell history +
    /// process listings.
    SetToken {
        /// Logical agent name (the container/agent name).
        agent: String,
        /// The PAT value inline. Mutually exclusive with `--token-stdin`.
        #[arg(long)]
        token: Option<String>,
        /// Read the PAT from stdin (trailing newline stripped). Mutually
        /// exclusive with `--token`.
        #[arg(long, conflicts_with = "token")]
        token_stdin: bool,
    },
}

// Default htpasswd file path — the host-side location of the gateway's
// credential store, pre-created by a tmpfiles rule when
// `services.hyperhive.gateway.auth.enable = true`. Literal lives in
// `hive_c0re::paths`.
use hive_c0re::paths::GATEWAY_HTPASSWD as DEFAULT_HTPASSWD_FILE;

#[derive(Subcommand)]
enum GatewayCmd {
    /// Add a new user or update the password of an existing user in the
    /// gateway htpasswd file. The password is hashed with `BCrypt` (cost 12).
    ///
    /// Pass `--password-stdin` when scripting or when you don't want the
    /// password visible in shell history. The file is created if it does
    /// not exist; its parent directory must already exist.
    CreateUser {
        /// Username to add or update.
        username: String,
        /// Set the password inline. WARNING: visible in shell history and
        /// process listings — prefer `--password-stdin` for sensitive input.
        /// Mutually exclusive with `--password-stdin`.
        #[arg(long, conflicts_with = "password_stdin")]
        password: Option<String>,
        /// Read the password from stdin (single line, trailing newline
        /// stripped). Mutually exclusive with `--password`.
        #[arg(long)]
        password_stdin: bool,
        /// Path to the htpasswd file. Defaults to the standard gateway
        /// credential store at `/var/lib/hyperhive/gateway/gateway.htpasswd`.
        #[arg(long, short = 'f', default_value = DEFAULT_HTPASSWD_FILE)]
        file: PathBuf,
    },
    /// Remove a user from the gateway htpasswd file. Exits with an error
    /// when the user is not found so callers can detect the no-op case.
    DeleteUser {
        /// Username to remove.
        username: String,
        /// Path to the htpasswd file. Defaults to the standard gateway
        /// credential store.
        #[arg(long, short = 'f', default_value = DEFAULT_HTPASSWD_FILE)]
        file: PathBuf,
    },
    /// List all usernames in the gateway htpasswd file, one per line.
    ListUsers {
        /// Path to the htpasswd file. Defaults to the standard gateway
        /// credential store.
        #[arg(long, short = 'f', default_value = DEFAULT_HTPASSWD_FILE)]
        file: PathBuf,
    },
}

#[derive(Subcommand)]
enum WgCmd {
    /// Generate (if absent) this hive's WireGuard private key, print its
    /// public key, and print the nix snippet to enable the mesh. Idempotent:
    /// an existing key is reused, never clobbered (clobbering would break a
    /// live mesh). Share the printed public key with peer hives.
    Init {
        /// This hive's mesh address (e.g. `10.42.0.1/32`) to bake into the
        /// printed snippet. Omit to get a placeholder you fill in.
        #[arg(long)]
        address: Option<String>,
    },
    /// Print the nix snippet to add a peer hive to the mesh. Pure output —
    /// paste it into this hive's config. Get `<pubkey>` from the peer's
    /// `hivectl wg init`.
    Peer {
        /// Peer hive's DNS domain (the `swarm.peers` attrset key).
        domain: String,
        /// Peer's WireGuard public key (from its `hivectl wg init`).
        #[arg(long)]
        pubkey: String,
        /// Peer's mesh address (e.g. `10.42.0.2/32`).
        #[arg(long)]
        address: String,
        /// Peer's `host:port` endpoint (omit for a peer that only dials out,
        /// e.g. one behind NAT — it must set an endpoint pointing back here).
        #[arg(long)]
        endpoint: Option<String>,
    },
    /// Show the live mesh interface state (`wg show wg-hive`). Requires the
    /// mesh to be enabled + up.
    Status,
}

#[derive(Subcommand)]
enum QuotaCmd {
    /// Enable btrfs qgroup accounting on the agent-state filesystem. Run
    /// once before `show` / `limit`. Triggers a full btrfs rescan (I/O
    /// heavy on a large filesystem), so it's a deliberate opt-in.
    /// Idempotent; a no-op on non-btrfs hosts.
    Enable,
    /// Report per-agent disk usage (referenced + exclusive bytes) from
    /// btrfs qgroups. With no name, shows every agent that has a state
    /// subvolume; pass a name to show just that one. Requires `enable`
    /// first.
    Show {
        /// Agent to show (omit for all agents with a state subvolume).
        name: Option<String>,
    },
    /// Set or clear an agent's disk quota (a referenced-usage cap). `size`
    /// accepts a byte count or a `K`/`M`/`G`/`T` suffix (e.g. `5G`), or
    /// `none` to clear the limit. Requires `enable` first.
    Limit {
        /// Agent whose state subvolume to limit.
        name: String,
        /// Size cap (`5G`, `500M`, `1073741824`) or `none` to clear.
        size: String,
    },
}

// Default host admin socket path. Shared with `hive-c0re`'s `main.rs`
// default via `hive_c0re::paths::HOST_SOCKET` — the daemon binds there
// and `hivectl agents` connects to it.
use hive_c0re::paths::HOST_SOCKET as DEFAULT_HOST_SOCKET;

#[derive(Subcommand)]
enum AgentsCmd {
    /// Show all managed agents with their status (running / needs-login /
    /// needs-update) and technical state (deployed sha, parent, pending
    /// reminders). The host roster overview; reuses the dashboard's
    /// per-agent aggregation. Requires the daemon running.
    List {
        /// Emit the raw JSON rows instead of the padded table (for
        /// scripting). The table is the default human-readable shape.
        #[arg(long)]
        json: bool,
    },
    /// Stop and start a single agent container without rebuilding config.
    /// Useful for "kick the container" when the process is stuck or the
    /// container needs a clean restart without changing the NixOS config.
    /// Rides the job queue (serialized against in-flight rebuilds for
    /// the same agent); waits with live progress unless `--no-wait`.
    Restart {
        /// Agent name (e.g. `damocles`, `ruth`).
        name: String,
        /// Return immediately after the restart DAG is queued.
        #[arg(long)]
        no_wait: bool,
    },
    /// Restart ALL managed agent containers via one restart DAG each —
    /// unrelated agents overlap, each serializes on its own lease.
    /// Waits for the whole set with live progress unless `--no-wait`.
    RestartAll {
        /// Return immediately after the restart DAGs are queued.
        #[arg(long)]
        no_wait: bool,
    },
}

#[derive(Subcommand)]
enum SubvolCmd {
    /// Convert an existing plain-dir agent state root into a btrfs subvolume
    /// in place. Stops the agent (so its state bind-mount is released),
    /// migrates `…/agents/<name>/` to a subvolume preserving
    /// ownership/permissions/xattrs, then restarts it. Idempotent (no-op if
    /// already a subvolume) and safe (the original dir is left untouched on
    /// any failure before the final swap). Requires `--yes` since it bounces
    /// the agent and moves its state.
    Upgrade {
        /// Agent name (e.g. `damocles`, `iris`).
        name: String,
        /// Confirm: this stops the agent, migrates its state dir, and
        /// restarts it. Required — the command refuses without it.
        #[arg(long)]
        yes: bool,
    },
    /// Read-only snapshots of an agent's state subvolume — the first step
    /// of the (in-progress) inter-hive migration path, or a manual
    /// point-in-time backup.
    Snapshot {
        #[command(subcommand)]
        cmd: SnapshotCmd,
    },
}

#[derive(Subcommand)]
enum SnapshotCmd {
    /// Create a read-only snapshot. Agent must already be a subvolume
    /// (`subvol upgrade` first). Prints the snapshot's host path.
    Create {
        /// Agent name (e.g. `damocles`, `iris`).
        name: String,
        /// Snapshot label. Mandatory, and must start with `hive-` — the
        /// prefix doubles as an allow-list hive-priv checks so only
        /// hivectl-issued snapshot names can reach the `btrfs subvolume
        /// snapshot` shellout.
        #[arg(long)]
        label: String,
    },
    /// Delete a snapshot created by `subvol snapshot create`.
    Delete {
        /// Agent name the snapshot belongs to.
        name: String,
        /// Snapshot label passed to `subvol snapshot create --label`.
        label: String,
    },
    /// Export a snapshot to a local file via `btrfs send` (the local-file
    /// half of inter-hive migration transport; the cross-hive `ssh ...
    /// btrfs receive` leg isn't wired up yet). Also useful standalone as a
    /// point-in-time backup: a full send with no `--parent` produces a
    /// self-contained archive of the snapshot.
    Send {
        /// Agent name the snapshot belongs to.
        name: String,
        /// Snapshot label passed to `subvol snapshot create --label`.
        label: String,
        /// Optional parent snapshot label for an incremental send
        /// (`btrfs send -p`) — must be an existing, older snapshot of the
        /// same agent. Omit for a full send.
        #[arg(long)]
        parent: Option<String>,
        /// Destination filename (not a path) under the migrate-staging
        /// dir. Refused if it already exists.
        #[arg(long)]
        dest: String,
    },
}

#[tokio::main]
async fn main() -> Result<()> {
    tracing_subscriber::fmt()
        .with_env_filter(
            tracing_subscriber::EnvFilter::try_from_default_env()
                .unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
        )
        .init();
    let cli = Cli::parse();
    let socket = cli.socket;
    match cli.cmd {
        Cmd::Forge { cmd } => match cmd {
            ForgeCmd::CreateUser {
                name,
                password,
                password_stdin,
            } => forge_create_user(&socket, &name, password.as_deref(), password_stdin).await,
        },
        Cmd::Matrix { cmd } => run_matrix_cmd(&socket, cmd).await,
        Cmd::Github { cmd } => match cmd {
            GithubCmd::SetToken {
                agent,
                token,
                token_stdin,
            } => github_set_token(&agent, token, token_stdin).await,
        },
        Cmd::Gateway { cmd } => match cmd {
            GatewayCmd::CreateUser {
                file,
                username,
                password,
                password_stdin,
            } => gateway_create_user(&file, &username, password.as_deref(), password_stdin),
            GatewayCmd::DeleteUser { file, username } => gateway_delete_user(&file, &username),
            GatewayCmd::ListUsers { file } => gateway_list_users(&file),
        },
        Cmd::Agents { cmd } => match cmd {
            AgentsCmd::List { json } => agents_list(&socket, json).await,
            AgentsCmd::Restart { name, no_wait } => agents_restart(&socket, &name, no_wait).await,
            AgentsCmd::RestartAll { no_wait } => agents_restart_all(&socket, no_wait).await,
        },
        Cmd::Wg { cmd } => match cmd {
            WgCmd::Init { address } => wg_init(&socket, address.as_deref()).await,
            WgCmd::Peer {
                domain,
                pubkey,
                address,
                endpoint,
            } => {
                wg_peer(&domain, &pubkey, &address, endpoint.as_deref());
                Ok(())
            }
            WgCmd::Status => wg_status(),
        },
        Cmd::PeerConfig {
            wg_address,
            wg_endpoint,
        } => {
            let domain = require_hive_domain(&socket).await?;
            peer_config(&domain, wg_address.as_deref(), wg_endpoint.as_deref());
            Ok(())
        }
        Cmd::Stop {
            scope,
            graceful,
            no_wait,
        } => stop(&socket, scope.to_scope(), graceful, no_wait).await,
        Cmd::Start { scope, no_wait } => start(&socket, scope.to_scope(), no_wait).await,
        Cmd::Restart { scope, graceful } => restart(&socket, scope.to_scope(), graceful).await,
        Cmd::Subvol { cmd } => match cmd {
            SubvolCmd::Upgrade { name, yes } => subvol_upgrade(&socket, &name, yes).await,
            SubvolCmd::Snapshot { cmd } => match cmd {
                SnapshotCmd::Create { name, label } => subvol_snapshot_create(&name, label).await,
                SnapshotCmd::Delete { name, label } => subvol_snapshot_delete(&name, &label).await,
                SnapshotCmd::Send {
                    name,
                    label,
                    parent,
                    dest,
                } => subvol_snapshot_send(&name, &label, parent.as_deref(), &dest).await,
            },
        },
        Cmd::Choom {
            name,
            resume_session,
        } => choom(&name, resume_session.as_deref()),
        Cmd::Quota { cmd } => match cmd {
            QuotaCmd::Enable => quota_enable().await,
            QuotaCmd::Show { name } => quota_show(name.as_deref()).await,
            QuotaCmd::Limit { name, size } => quota_limit(&name, &size).await,
        },
        Cmd::MarkdownDocs => {
            print!("{}", clap_markdown::help_markdown::<Cli>());
            Ok(())
        }
        Cmd::Open { target } => open_url(&socket, target).await,
        Cmd::Completions { shell } => {
            generate_completions(shell);
            Ok(())
        }
    }
}

/// Route a `matrix` subcommand to its handler. Extracted from `main`'s
/// dispatch match so the top-level router stays small.
async fn run_matrix_cmd(socket: &Path, cmd: MatrixCmd) -> Result<()> {
    match cmd {
        MatrixCmd::CreateUser {
            name,
            password,
            password_stdin,
        } => matrix_create_user(socket, &name, password.as_deref(), password_stdin).await,
        MatrixCmd::SyncAdmin => matrix_sync_admin(socket).await,
        MatrixCmd::PromoteUser { name } => matrix_promote_user(socket, &name).await,
        MatrixCmd::ResetPassword { name } => matrix_reset_password(socket, &name).await,
        MatrixCmd::Invite { user, room } => matrix_invite(socket, &user, room.as_deref()).await,
    }
}

/// `open <home|forge|matrix>` — resolve the surface URL from the daemon,
/// print it, then best-effort `xdg-open` it. Printing is the reliable
/// core (headless / SSH hosts where no browser opener exists); the open
/// is convenience on top, so a missing/failed `xdg-open` is not an error.
async fn open_url(socket: &Path, target: OpenTarget) -> Result<()> {
    let urls = query_hive_urls(socket).await.with_context(|| {
        format!(
            "could not reach the hive-c0re daemon for URLs — is hive-c0re running? \
             (the socket is at {DEFAULT_HOST_SOCKET})"
        )
    })?;
    let (url, hint) = match target {
        OpenTarget::Home => (
            urls.home,
            "the dashboard URL needs `services.hyperhive.domain` to be set",
        ),
        OpenTarget::Forge => (
            urls.forge,
            "the public forge URL needs `services.hyperhive.forge.behindGateway = true`",
        ),
        OpenTarget::Matrix => (
            urls.matrix,
            "the matrix GUI URL needs `services.hyperhive.matrix.gui.enable = true`",
        ),
    };
    let url = url.with_context(|| format!("no URL available for this surface — {hint}"))?;
    println!("{url}");
    // Best-effort: many hosts are headless, so a missing opener or a
    // non-zero exit is fine — the URL is already printed.
    match std::process::Command::new("xdg-open").arg(&url).status() {
        Ok(status) if status.success() => {}
        Ok(status) => eprintln!("note: xdg-open exited with {status} (URL printed above)"),
        Err(e) => eprintln!("note: could not run xdg-open ({e}) (URL printed above)"),
    }
    Ok(())
}

/// Emit a shell completion script for `hivectl` to stdout. Walks the clap
/// command tree (the single source of truth — same tree `markdown-docs`
/// renders) so completions never drift from the actual verbs/flags.
fn generate_completions(shell: clap_complete::Shell) {
    use clap::CommandFactory as _;
    let mut cmd = Cli::command();
    clap_complete::generate(shell, &mut cmd, "hivectl", &mut std::io::stdout());
}

/// Host path of this hive's WireGuard private key (matches the
/// `privateKeyFile` example in the swarm.wireguard nix options).
const WG_KEY_PATH: &str = "/etc/wireguard/hive.key";
/// The mesh interface name hive-c0re's nix module brings up.
const WG_INTERFACE: &str = "wg-hive";

/// Host path of this hive's self-signed CA cert (matches the
/// `services.hyperhive.tls.stateDir` default in hive-tls.nix). Its
/// existence means the gateway serves a self-signed, hive-CA-signed leaf,
/// so a federating peer needs this CA via `swarm.peers.<d>.caCert`. Absent
/// = ACME / operator cert (trusted by the default CA bundle, no `caCert`).
const HIVE_TLS_CA_PATH: &str = "/var/lib/hive-tls/ca.pem";

/// Best-effort query for this hive's domain from the running daemon
/// (`HostRequest::Urls`, which reads `HYPERHIVE_HIVE_DOMAIN` from c0re's
/// service env). `None` when the daemon is unreachable or the domain is
/// unset — callers decide whether that's fatal.
async fn query_hive_domain(socket: &Path) -> Option<String> {
    query_hive_urls(socket).await.and_then(|u| u.domain)
}

/// Best-effort query for this hive's domain + browser-facing web URLs
/// (`HostRequest::Urls`). `None` when the daemon is unreachable.
async fn query_hive_urls(socket: &Path) -> Option<hive_host_sock::HiveUrls> {
    hive_c0re::client::request(socket, hive_host_sock::HostRequest::Urls)
        .await
        .ok()
        .and_then(|r| r.urls)
}

/// Require this hive's domain from the daemon for snippet generation.
/// Errors with a clear hint when it can't be resolved, so `peer-config`
/// never silently emits a wrong key.
async fn require_hive_domain(socket: &Path) -> Result<String> {
    query_hive_domain(socket).await.context(
        "could not determine this hive's domain from the daemon — is hive-c0re running \
         and `services.hyperhive.domain` set?",
    )
}

/// `wg init` — generate (if absent) the hive's WireGuard key, print its
/// public key + the nix snippet to enable the mesh, then (best-effort)
/// the `peer-config` block peers paste to federate with this hive, so a
/// fresh setup is one command. The domain comes from the daemon; if it
/// can't be resolved, the peer block is skipped (init still succeeds —
/// its core job is enabling the mesh locally).
async fn wg_init(socket: &Path, address: Option<&str>) -> Result<()> {
    use std::os::unix::fs::PermissionsExt as _;
    let key_path = Path::new(WG_KEY_PATH);
    if key_path.exists() {
        println!(
            "WireGuard private key already present at {WG_KEY_PATH} — reusing (not regenerating)."
        );
    } else {
        if let Some(parent) = key_path.parent() {
            std::fs::create_dir_all(parent)
                .with_context(|| format!("create {}", parent.display()))?;
            std::fs::set_permissions(parent, std::fs::Permissions::from_mode(0o700)).ok();
        }
        let out = std::process::Command::new("wg")
            .arg("genkey")
            .output()
            .context("run `wg genkey` (is wireguard-tools installed?)")?;
        if !out.status.success() {
            bail!(
                "wg genkey failed: {}",
                String::from_utf8_lossy(&out.stderr).trim()
            );
        }
        std::fs::write(key_path, &out.stdout)
            .with_context(|| format!("write {}", key_path.display()))?;
        std::fs::set_permissions(key_path, std::fs::Permissions::from_mode(0o400))
            .with_context(|| format!("chmod 400 {}", key_path.display()))?;
        println!("Generated WireGuard private key at {WG_KEY_PATH} (0400).");
    }

    let privkey =
        std::fs::read(key_path).with_context(|| format!("read {}", key_path.display()))?;
    let pubkey = wg_pubkey(&privkey)?;
    let addr = address.unwrap_or("<MESH_ADDRESS e.g. 10.42.0.1/32>");
    println!("\nPublic key (share this with peer hives — they pass it to `hivectl wg peer`):");
    println!("  {pubkey}");
    println!("\nAdd to this hive's NixOS config:");
    println!("  services.hyperhive.swarm.wireguard = {{");
    println!("    enable = true;");
    println!("    privateKeyFile = \"{WG_KEY_PATH}\";");
    println!("    address = \"{addr}\";");
    println!("    # listenPort = 51820;  # default");
    println!("  }};");

    // Also print the block a peer pastes to federate with us (CA + this
    // mesh key) — one-stop setup. Domain comes from the daemon;
    // best-effort, so init still succeeds when it can't be resolved.
    if let Some(d) = query_hive_domain(socket).await {
        println!();
        peer_config(&d, address, None);
    }
    Ok(())
}

/// Derive a WireGuard public key from a private key by piping it through
/// `wg pubkey`.
fn wg_pubkey(privkey: &[u8]) -> Result<String> {
    use std::io::Write as _;
    use std::process::{Command, Stdio};
    let mut child = Command::new("wg")
        .arg("pubkey")
        .stdin(Stdio::piped())
        .stdout(Stdio::piped())
        .stderr(Stdio::piped())
        .spawn()
        .context("run `wg pubkey` (is wireguard-tools installed?)")?;
    child
        .stdin
        .take()
        .context("wg pubkey: stdin unavailable")?
        .write_all(privkey)
        .context("write private key to `wg pubkey`")?;
    let out = child.wait_with_output().context("wait for `wg pubkey`")?;
    if !out.status.success() {
        bail!(
            "wg pubkey failed: {}",
            String::from_utf8_lossy(&out.stderr).trim()
        );
    }
    Ok(String::from_utf8_lossy(&out.stdout).trim().to_owned())
}

/// `wg peer` — print the nix snippet to add a peer hive to the mesh.
/// Pure output (no fallible work), so it returns `()`; the dispatch arm
/// wraps it in `Ok` to match the sibling verbs.
fn wg_peer(domain: &str, pubkey: &str, address: &str, endpoint: Option<&str>) {
    println!("Add to this hive's NixOS config:");
    println!("  services.hyperhive.swarm.peers.\"{domain}\" = {{");
    println!("    wireguardPublicKey = \"{pubkey}\";");
    println!("    wireguardAddress = \"{address}\";");
    if let Some(ep) = endpoint {
        println!("    wireguardEndpoint = \"{ep}\";");
    }
    println!("  }};");
}

/// `peer-config` — print the `swarm.peers."<domain>"` block a peer
/// operator pastes to federate with THIS hive, plus a `cp` line for the
/// CA when this hive is self-signed. Reads local state only (the TLS CA
/// cert presence + the wg key); prints, never mutates.
fn peer_config(domain: &str, wg_address: Option<&str>, wg_endpoint: Option<&str>) {
    let self_signed = Path::new(HIVE_TLS_CA_PATH).exists();
    // CA filename derived from the first DNS label so multiple peers'
    // certs don't collide in the operator's config dir.
    let ca_file = format!("{}-ca.pem", domain.split('.').next().unwrap_or("peer"));
    // WireGuard public key, when this hive has a mesh key. Best-effort:
    // a missing key or absent `wg` binary just omits the mesh lines.
    let wg_pub = std::fs::read(WG_KEY_PATH)
        .ok()
        .and_then(|k| wg_pubkey(&k).ok());

    if self_signed {
        println!("# 1. copy this hive's CA cert next to the peer's config:");
        println!("cp {HIVE_TLS_CA_PATH} ./{ca_file}");
        println!();
        println!("# 2. paste into the peer hive's NixOS config:");
    } else {
        println!("# paste into the peer hive's NixOS config:");
    }
    println!("services.hyperhive.swarm.peers.\"{domain}\" = {{");
    if self_signed {
        println!("  caCert = ./{ca_file};");
    }
    if let Some(pk) = &wg_pub {
        println!("  wireguardPublicKey = \"{pk}\";");
    }
    if let Some(addr) = wg_address {
        println!("  wireguardAddress = \"{addr}\";");
    }
    if let Some(ep) = wg_endpoint {
        println!("  wireguardEndpoint = \"{ep}\";");
    }
    println!("}};");
    if !self_signed {
        println!(
            "# (this hive's cert chains to a public CA — no `caCert` needed; \
             it's trusted by the default bundle.)"
        );
    }
}

/// `wg status` — show the live mesh interface (`wg show wg-hive`),
/// inheriting stdout so the operator sees it directly.
fn wg_status() -> Result<()> {
    let status = std::process::Command::new("wg")
        .args(["show", WG_INTERFACE])
        .status()
        .context("run `wg show` (is wireguard-tools installed?)")?;
    if !status.success() {
        bail!(
            "`wg show {WG_INTERFACE}` failed — is the mesh enabled + up? \
             (services.hyperhive.swarm.wireguard.enable = true, then deploy)"
        );
    }
    Ok(())
}

/// `quota enable` — turn on btrfs qgroup accounting (operator opt-in).
async fn quota_enable() -> Result<()> {
    hive_c0re::priv_client::ensure_btrfs_quota()
        .await
        .context("enable btrfs qgroup accounting")?;
    println!("btrfs qgroup accounting enabled on the agent-state filesystem.");
    println!("(usage may read 0 until btrfs finishes its background rescan)");
    Ok(())
}

/// `quota show [name]` — report per-agent disk usage from btrfs qgroups.
async fn quota_show(name: Option<&str>) -> Result<()> {
    let agents: Vec<String> = match name {
        Some(n) => vec![n.to_owned()],
        None => Coordinator::kept_state_names(),
    };
    if agents.is_empty() {
        println!("no agents with a state dir found");
        return Ok(());
    }
    for agent in &agents {
        match hive_c0re::priv_client::read_subvolume_usage(agent).await {
            Ok((rfer, excl)) => {
                println!(
                    "{agent:<12} referenced {:>10}   exclusive {:>10}",
                    human_bytes(rfer),
                    human_bytes(excl)
                );
            }
            Err(e) => {
                let msg = format!("{e:#}");
                // btrfs-progs prints "ERROR: ... quota not enabled" to stderr
                // when qgroups are off; match the stable fragment case-
                // insensitively rather than an exact line (the surrounding
                // wording varies across btrfs-progs versions).
                if msg.to_ascii_lowercase().contains("quota not enabled") {
                    bail!("btrfs quota not enabled — run `hivectl quota enable` first");
                }
                // A plain-dir agent (no subvolume) has no qgroup; report it
                // inline and keep going rather than aborting the whole sweep.
                println!("{agent:<12} (no qgroup data — plain dir or: {msg})");
            }
        }
    }
    Ok(())
}

/// `quota limit <name> <size>` — set or clear an agent's disk quota.
async fn quota_limit(name: &str, size: &str) -> Result<()> {
    let limit = parse_quota_size(size)?;
    hive_c0re::priv_client::set_subvolume_quota(name, limit)
        .await
        .with_context(|| format!("set quota for {name}"))?;
    match limit {
        Some(n) => println!("set {name} quota to {} ({n} bytes)", human_bytes(n)),
        None => println!("cleared {name} quota"),
    }
    Ok(())
}

/// Parse a quota size: a byte count, a `K`/`M`/`G`/`T`-suffixed value
/// (powers of 1024), or `none` to clear. Rejects overflow + bad input.
fn parse_quota_size(s: &str) -> Result<Option<u64>> {
    let t = s.trim();
    if t.eq_ignore_ascii_case("none") {
        return Ok(None);
    }
    let (num, mult) = match t.chars().last() {
        Some('K' | 'k') => (&t[..t.len() - 1], 1024u64),
        Some('M' | 'm') => (&t[..t.len() - 1], 1024 * 1024),
        Some('G' | 'g') => (&t[..t.len() - 1], 1024 * 1024 * 1024),
        Some('T' | 't') => (&t[..t.len() - 1], 1024u64 * 1024 * 1024 * 1024),
        _ => (t, 1),
    };
    let n: u64 = num.trim().parse().with_context(|| {
        format!("invalid size {s:?} — use a byte count, a K/M/G/T suffix (e.g. 5G), or `none`")
    })?;
    n.checked_mul(mult)
        .map(Some)
        .with_context(|| format!("size {s:?} overflows u64"))
}

/// Format a byte count as a human-friendly KiB/MiB/GiB/TiB string.
#[allow(clippy::cast_precision_loss)] // display-only; precision loss is cosmetic
fn human_bytes(n: u64) -> String {
    const UNITS: [&str; 5] = ["B", "KiB", "MiB", "GiB", "TiB"];
    if n < 1024 {
        return format!("{n} B");
    }
    let mut v = n as f64;
    let mut i = 0;
    while v >= 1024.0 && i < UNITS.len() - 1 {
        v /= 1024.0;
        i += 1;
    }
    format!("{v:.1} {}", UNITS[i])
}

/// True when `name` matches an existing hyperhive agent — i.e. it has a
/// persistent state dir under `/var/lib/hyperhive/agents/`. We use the
/// state dir (not the live container list) so kept-state tombstones
/// still resolve as agents — re-provisioning a destroyed-but-kept agent
/// should still drop its token in the existing state tree.
///
/// Uses `try_exists()` rather than `Path::exists()` so a permission
/// error reaching the agents root is surfaced, not collapsed into
/// `false`. The agents root is `0700 hive-core`, so running hivectl
/// without root yields EACCES on traversal — `Path::exists()` would
/// silently report `false`, which callers turn into a misleading "no
/// such agent" (or, for the create-user paths, a silent misclassify of
/// a real agent as a non-agent account). Mapping EACCES to an explicit
/// "needs root" error fixes that first-run footgun, where running a
/// privileged verb without sudo reported as a missing agent.
fn agent_exists(name: &str) -> Result<bool> {
    let root = hive_c0re::paths::agent_state_dir(name);
    match root.try_exists() {
        Ok(found) => Ok(found),
        Err(e) if e.kind() == std::io::ErrorKind::PermissionDenied => bail!(
            "cannot read the agents root {} (permission denied) - this command needs root; \
             re-run with sudo",
            root.parent().unwrap_or(&root).display()
        ),
        Err(e) => Err(e).with_context(|| format!("check agent state dir {}", root.display())),
    }
}

/// Drop into an interactive Claude session in the agent container.
///
/// Execs `machinectl shell <name>@h-<name>` running claude as the agent
/// user from its state dir, reproducing the harness's per-turn claude
/// invocation (flags, session selection, why it never collides with the
/// live harness session). See `docs/tools/hivectl.md` (Choom) for the
/// full rationale. Inherits the caller's PTY; requires root + a running
/// container.
fn choom(name: &str, resume_session: Option<&str>) -> Result<()> {
    if !agent_exists(name)? {
        bail!(
            "no such agent: '{name}' (no state dir under {}/)",
            hive_c0re::paths::AGENTS_ROOT
        );
    }
    let container = hive_c0re::lifecycle::container_name(name);
    // Enter as the agent's unix user (== agent name) so claude reads the
    // right `$HOME/.claude`.
    let target = format!("{name}@{container}");
    let claude = "/run/current-system/sw/bin/claude";
    // Bind-mounted state dir; matches `hive-ag3nt::paths::state_dir()`.
    let state_dir = format!("/agents/{name}/state");
    // Per-turn config the harness writes; matches `paths::config_dir()`.
    let cfg = "/run/hive-config";
    // `--resume <value>` passes through as `claude --resume <value>` —
    // the rejoin-by-id surface. (claude's `--continue` is a bare flag
    // that resumes the cwd's latest session — the harness's — and would
    // consume a trailing value as the first PROMPT; choom never uses
    // it.) The value is single-quoted into the shell script, so reject
    // an embedded single quote (the only char that breaks
    // single-quoting) to rule out injection — session ids never
    // contain one.
    let session_arg = match resume_session {
        Some(val) => {
            if val.contains('\'') {
                bail!("invalid --resume value '{val}': must not contain a single quote");
            }
            format!("set -- --resume '{val}';")
        }
        None => "set --;".to_string(),
    };
    // Build claude's argv as the harness does, each flag included only
    // when its file is present so a half-up container degrades to a bare
    // session. `name` is `[a-z0-9._-]` so there's nothing to quote.
    let inner = format!(
        "cd {state_dir} || exit 1; {session_arg} \
         [ -f {cfg}/claude-settings.json ] && set -- \"$@\" --settings {cfg}/claude-settings.json; \
         [ -f {cfg}/claude-mcp-config.json ] && set -- \"$@\" --mcp-config {cfg}/claude-mcp-config.json; \
         [ -f {cfg}/claude-system-prompt.md ] && set -- \"$@\" --system-prompt-file {cfg}/claude-system-prompt.md; \
         exec {claude} \"$@\""
    );
    let mut cmd = std::process::Command::new("machinectl");
    cmd.arg("shell")
        .arg(&target)
        .arg("/bin/sh")
        .arg("-lc")
        .arg(&inner);
    // exec() replaces the current process — we inherit stdin/stdout/stderr
    // (the caller's PTY) so the Claude session is fully interactive.
    // This call only returns on error.
    let err = cmd.exec();
    Err(anyhow::anyhow!("exec machinectl: {err}"))
}

/// `hivectl github set-token <agent>`: write an operator-supplied GitHub PAT
/// into the agent's `github-token` state file (0600, agent-owned) via
/// hive-priv, so the agent's `gh` wrapper + git credential helper can
/// authenticate. Read live at invocation, so no rebuild/restart is needed.
async fn github_set_token(agent: &str, token: Option<String>, token_stdin: bool) -> Result<()> {
    let token = match (token, token_stdin) {
        (Some(t), _) => t,
        (None, true) => {
            let mut s = String::new();
            std::io::Read::read_to_string(&mut std::io::stdin(), &mut s)?;
            s.trim_end_matches(['\n', '\r']).to_owned()
        }
        (None, false) => bail!(
            "provide the PAT via --token <pat> or --token-stdin (stdin preferred — \
             an inline token is visible in shell history + process listings)"
        ),
    };
    if token.is_empty() {
        bail!("refusing to write an empty GitHub token for agent '{agent}'");
    }
    hive_c0re::priv_client::write_agent_github_token(agent, &token).await?;
    println!(
        "wrote github-token for agent '{agent}' \
         (read live by the gh wrapper / git credential helper — no rebuild needed)"
    );
    Ok(())
}

async fn forge_create_user(
    socket: &Path,
    name: &str,
    password: Option<&str>,
    password_stdin: bool,
) -> Result<()> {
    // Resolve the password client-side (inline flag or stdin read); the
    // daemon never touches this process's stdin. The is-present check, the
    // agent-vs-operator branch, and token persistence now live in the
    // daemon handler.
    let password = resolve_password(password, password_stdin)?;
    forge_request(
        socket,
        hive_host_sock::HostRequest::ForgeCreateUser {
            name: name.to_owned(),
            password,
        },
    )
    .await
}

/// Send a forge provisioning request to the daemon and print its result
/// lines. Mirrors [`matrix_request`] — the daemon owns the provisioning
/// logic; hivectl just relays the outcome.
async fn forge_request(socket: &Path, req: hive_host_sock::HostRequest) -> Result<()> {
    let resp = hive_c0re::client::request(socket, req)
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !resp.ok {
        bail!(
            "forge: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    for line in &resp.messages {
        println!("{line}");
    }
    Ok(())
}

/// Resolve the password the caller asked for, or return `None` to fall
/// back to a random throwaway. `--password <PW>` wins outright;
/// `--password-stdin` reads one line off stdin (trailing newline
/// stripped). Empty stdin is treated as an error so the caller doesn't
/// silently provision an empty-password account.
fn resolve_password(password: Option<&str>, password_stdin: bool) -> Result<Option<String>> {
    if let Some(p) = password {
        return Ok(Some(p.to_owned()));
    }
    if password_stdin {
        use std::io::BufRead as _;
        let stdin = std::io::stdin();
        let mut line = String::new();
        stdin
            .lock()
            .read_line(&mut line)
            .context("read password from stdin")?;
        let trimmed = line.trim_end_matches(['\r', '\n']).to_owned();
        if trimmed.is_empty() {
            bail!("--password-stdin: empty input");
        }
        return Ok(Some(trimmed));
    }
    Ok(None)
}

/// Send a matrix provisioning request to the daemon and print the
/// operator-facing result lines it returns. The daemon owns the register +
/// admin tokens and the matrix creds dir, so hivectl no longer links the
/// matrix machinery — it just forwards the request and renders the reply.
async fn matrix_request(socket: &Path, req: hive_host_sock::HostRequest) -> Result<()> {
    let resp = hive_c0re::client::request(socket, req)
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !resp.ok {
        bail!(
            "matrix: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    for line in &resp.messages {
        println!("{line}");
    }
    Ok(())
}

async fn matrix_create_user(
    socket: &Path,
    name: &str,
    password: Option<&str>,
    password_stdin: bool,
) -> Result<()> {
    // Resolve the password client-side (an inline flag or a stdin read);
    // the daemon never touches this process's stdin. The agent-vs-operator
    // branch + throwaway-password handling now live in the daemon handler.
    let password = resolve_password(password, password_stdin)?;
    matrix_request(
        socket,
        hive_host_sock::HostRequest::MatrixCreateUser {
            name: name.to_owned(),
            password,
        },
    )
    .await
}

async fn matrix_sync_admin(socket: &Path) -> Result<()> {
    matrix_request(socket, hive_host_sock::HostRequest::MatrixSyncAdmin).await
}

async fn matrix_promote_user(socket: &Path, name: &str) -> Result<()> {
    matrix_request(
        socket,
        hive_host_sock::HostRequest::MatrixPromoteUser {
            name: name.to_owned(),
        },
    )
    .await
}

async fn matrix_invite(socket: &Path, user: &str, room: Option<&str>) -> Result<()> {
    matrix_request(
        socket,
        hive_host_sock::HostRequest::MatrixInvite {
            user: user.to_owned(),
            room: room.map(str::to_owned),
        },
    )
    .await
}

async fn matrix_reset_password(socket: &Path, name: &str) -> Result<()> {
    matrix_request(
        socket,
        hive_host_sock::HostRequest::MatrixResetPassword {
            name: name.to_owned(),
        },
    )
    .await
}

// ---------------------------------------------------------------------------
// Gateway htpasswd helpers
// ---------------------------------------------------------------------------

/// Read an htpasswd file into a list of lines, or return an empty list
/// if the file does not exist yet.
fn htpasswd_read(path: &Path) -> Result<Vec<String>> {
    if !path.exists() {
        return Ok(vec![]);
    }
    let content = std::fs::read_to_string(path)
        .with_context(|| format!("read htpasswd file {}", path.display()))?;
    Ok(content.lines().map(str::to_owned).collect())
}

/// Write lines back to `path` atomically (write to `<path>.tmp`, then
/// rename). A trailing newline is always appended to the last line.
fn htpasswd_write(path: &Path, lines: &[String]) -> Result<()> {
    let tmp = path.with_extension("htpasswd.tmp");
    let content = if lines.is_empty() {
        String::new()
    } else {
        let mut s = lines.join("\n");
        s.push('\n');
        s
    };
    std::fs::write(&tmp, &content)
        .with_context(|| format!("write htpasswd tmp {}", tmp.display()))?;
    std::fs::rename(&tmp, path)
        .with_context(|| format!("rename {} → {}", tmp.display(), path.display()))?;
    Ok(())
}

/// Add or update `username` in the htpasswd file at `file`, hashing
/// `password` with `BCrypt` (cost 12). Creates the file when absent.
fn gateway_create_user(
    file: &Path,
    username: &str,
    password: Option<&str>,
    password_stdin: bool,
) -> Result<()> {
    let pw = resolve_password(password, password_stdin)?.ok_or_else(|| {
        anyhow::anyhow!("a password is required — pass --password or --password-stdin")
    })?;
    validate_htpasswd_username(username)?;
    let raw_hash = bcrypt::hash(&pw, 12).context("bcrypt hash")?;
    // nginx auth_basic only recognises $2a$/$2x$/$2y$ — not $2b$. The two
    // prefixes are algorithmically identical; remap so nginx accepts the hash.
    let hash = raw_hash.replacen("$2b$", "$2y$", 1);
    let entry = format!("{username}:{hash}");
    let mut lines = htpasswd_read(file)?;
    let prefix = format!("{username}:");
    if let Some(pos) = lines.iter().position(|l| l.starts_with(&prefix)) {
        lines[pos] = entry;
        htpasswd_write(file, &lines)?;
        println!(
            "gateway: updated password for '{username}' in {}",
            file.display()
        );
    } else {
        lines.push(entry);
        htpasswd_write(file, &lines)?;
        println!("gateway: added user '{username}' to {}", file.display());
    }
    Ok(())
}

/// Remove `username` from the htpasswd file. Errors when the user is
/// not present so callers can detect the no-op case.
fn gateway_delete_user(file: &Path, username: &str) -> Result<()> {
    let mut lines = htpasswd_read(file)?;
    let prefix = format!("{username}:");
    let before = lines.len();
    lines.retain(|l| !l.starts_with(&prefix));
    if lines.len() == before {
        bail!("gateway: user '{username}' not found in {}", file.display());
    }
    htpasswd_write(file, &lines)?;
    println!("gateway: removed user '{username}' from {}", file.display());
    Ok(())
}

/// Print one username per line from the htpasswd file.
fn gateway_list_users(file: &Path) -> Result<()> {
    let lines = htpasswd_read(file)?;
    for line in &lines {
        // Skip blank lines and comments.
        if line.is_empty() || line.starts_with('#') {
            continue;
        }
        if let Some((name, _)) = line.split_once(':') {
            println!("{name}");
        }
    }
    Ok(())
}

// ---------------------------------------------------------------------------
// Agent management helpers (require daemon via host admin socket)
// ---------------------------------------------------------------------------

async fn agents_restart(socket: &Path, name: &str, no_wait: bool) -> Result<()> {
    let resp = hive_c0re::client::request(
        socket,
        hive_host_sock::HostRequest::Restart {
            name: name.to_owned(),
        },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if resp.ok {
        println!("restart queued: {name}");
        wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
    } else {
        bail!(
            "restart {name}: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        )
    }
}

/// `hivectl agents list` — fetch the per-agent status roster from the
/// daemon (`HostRequest::AgentStatus`) and render it as a padded table,
/// or the raw JSON rows with `--json`. Reuses the dashboard's
/// `ContainerView` aggregation, so the CLI and the web UI never drift.
async fn agents_list(socket: &Path, json: bool) -> Result<()> {
    let resp = hive_c0re::client::request(socket, hive_host_sock::HostRequest::AgentStatus)
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !resp.ok {
        bail!(
            "agents list: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    let rows = resp.agent_statuses.unwrap_or_default();
    if json {
        println!("{}", serde_json::to_string_pretty(&rows)?);
        return Ok(());
    }
    if rows.is_empty() {
        println!("no managed agents found");
        return Ok(());
    }
    // STATUS collapses the health flags into one space-separated token so
    // the common case (`running`) stays short and anomalies stand out.
    let status_of = |r: &hive_sh4re::AgentStatusRow| -> String {
        let mut s = if r.running { "running" } else { "stopped" }.to_owned();
        if r.needs_login {
            s.push_str(" needs-login");
        }
        if r.needs_update {
            s.push_str(" needs-update");
        }
        s
    };
    let headers = ["NAME", "STATUS", "REV", "PARENT", "REMIND"];
    let table: Vec<[String; 5]> = rows
        .iter()
        .map(|r| {
            [
                r.name.clone(),
                status_of(r),
                r.deployed_sha.clone().unwrap_or_else(|| "-".to_owned()),
                r.parent.clone().unwrap_or_else(|| "-".to_owned()),
                if r.pending_reminders > 0 {
                    r.pending_reminders.to_string()
                } else {
                    "-".to_owned()
                },
            ]
        })
        .collect();
    let mut widths = headers.map(str::len);
    for row in &table {
        for (i, cell) in row.iter().enumerate() {
            widths[i] = widths[i].max(cell.len());
        }
    }
    let fmt_row = |cells: &[String]| -> String {
        cells
            .iter()
            .enumerate()
            .map(|(i, c)| format!("{c:<w$}", w = widths[i]))
            .collect::<Vec<_>>()
            .join("  ")
            .trim_end()
            .to_owned()
    };
    let header_cells: Vec<String> = headers.iter().map(|h| (*h).to_owned()).collect();
    println!("{}", fmt_row(&header_cells));
    for row in &table {
        println!("{}", fmt_row(row));
    }
    Ok(())
}

async fn agents_restart_all(socket: &Path, no_wait: bool) -> Result<()> {
    let resp = hive_c0re::client::request(socket, hive_host_sock::HostRequest::RestartAll)
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    let agents = resp.agents.as_deref().unwrap_or(&[]);
    if agents.is_empty() {
        println!("restart-all: no managed containers found");
    } else {
        for a in agents {
            println!("restart queued: {a}");
        }
    }
    if !resp.ok {
        bail!(
            "restart-all: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
}

async fn stop(
    socket: &Path,
    scope: hive_host_sock::LifecycleScope,
    graceful: bool,
    no_wait: bool,
) -> Result<()> {
    let resp = hive_c0re::client::request(
        socket,
        hive_host_sock::HostRequest::Stop { scope, graceful },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    // Render first, but even when an infra failure makes it bail,
    // watch the already-queued agent DAGs before surfacing the error —
    // they run regardless.
    let rendered = render_lifecycle(&resp, "stop queued");
    wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?;
    rendered
}

async fn start(socket: &Path, scope: hive_host_sock::LifecycleScope, no_wait: bool) -> Result<()> {
    let resp = hive_c0re::client::request(socket, hive_host_sock::HostRequest::Start { scope })
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    let rendered = render_lifecycle(&resp, "start queued");
    wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), no_wait).await?;
    rendered
}

/// Restart — one `RestartScoped` daemon call, server-side DAG-based (see
/// issue tracker "dagify hivectl commands"). Each targeted agent rides
/// exactly one atomic DAG queued up front — `Restart` (mechanical stop +
/// reconcile), or `GracefulRestart` with `--graceful` (signal → drain →
/// mechanical stop → reconcile); infra containers restart synchronously.
/// No "submit one DAG, wait for it, submit another" composition on either
/// side of the wire: unlike the old client-side stop-then-start compose, a
/// dropped `hivectl` connection mid-restart no longer leaves an agent
/// stopped with no automatic follow-up — the daemon owns the whole
/// sequence once this call is made.
///
/// No `--no-wait` here on purpose, same as before: the operator wants to
/// see the restart actually land, not just get queued.
async fn restart(
    socket: &Path,
    scope: hive_host_sock::LifecycleScope,
    graceful: bool,
) -> Result<()> {
    let resp = hive_c0re::client::request(
        socket,
        hive_host_sock::HostRequest::RestartScoped { scope, graceful },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    let rendered = render_lifecycle(&resp, "restart queued");
    wait_for_dags(socket, resp.queued_dags.unwrap_or_default(), false).await?;
    rendered
}

/// A [`LifecycleScope`](hive_host_sock::LifecycleScope) targeting exactly one
/// agent by name (no infra containers, no all-agents flag).
fn single_agent_scope(name: &str) -> hive_host_sock::LifecycleScope {
    hive_host_sock::LifecycleScope {
        agents: false,
        agent_names: vec![name.to_owned()],
        ci: false,
        forge: false,
        gateway: false,
        matrix: false,
    }
}

/// `subvol upgrade <agent>` — migrate an existing plain-dir agent state root
/// to a btrfs subvolume. Composed client-side (like `restart`): stop the
/// agent so its state bind-mount is released, run the privileged in-place
/// migration via hive-priv, then restart it. The restart is attempted
/// regardless of the migration outcome so a failed migration never leaves
/// the agent down; the migration error (if any) is surfaced afterwards.
async fn subvol_upgrade(socket: &Path, name: &str, yes: bool) -> Result<()> {
    if !agent_exists(name)? {
        bail!("no agent named {name:?} (no state dir under the agents root)");
    }
    if !yes {
        bail!(
            "`subvol upgrade {name}` stops the agent, migrates its state dir to a btrfs \
             subvolume, then restarts it. Re-run with --yes to proceed."
        );
    }

    println!("stopping {name} (releasing its state bind-mount)…");
    let stop_resp = hive_c0re::client::request(
        socket,
        hive_host_sock::HostRequest::Stop {
            scope: single_agent_scope(name),
            graceful: false,
        },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !stop_resp.ok {
        bail!(
            "stop {name}: {}",
            stop_resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    // The stop is a queued DAG now — the migration below snapshots +
    // swaps the state dir and MUST NOT run under a live bind mount, so
    // wait for the stop to actually execute before touching anything.
    wait_for_dags(socket, stop_resp.queued_dags.unwrap_or_default(), false)
        .await
        .with_context(|| format!("waiting for {name} to stop before the migration"))?;

    println!("migrating {name} state dir to a btrfs subvolume…");
    let upgrade = hive_c0re::priv_client::upgrade_agent_subvolume(name).await;

    // Always attempt the restart, even if the migration failed — don't leave
    // the agent down. Capture the result rather than `?`-ing it so a
    // start-side failure (incl. the IPC call itself erroring) can't mask the
    // migration outcome below.
    println!("starting {name}…");
    let start_result = hive_c0re::client::request(
        socket,
        hive_host_sock::HostRequest::Start {
            scope: single_agent_scope(name),
        },
    )
    .await;

    // Surface the migration outcome FIRST — it's the meaningful result and
    // must not be shadowed by a restart-side failure. On migration failure the
    // original state dir is untouched (the priv op rolls back before the swap).
    upgrade.with_context(|| format!("upgrade {name} state subvolume"))?;

    // Migration succeeded; now surface any restart problem — either the IPC
    // call erroring, or the daemon reporting a failed start. The migration is
    // done regardless, so point at the manual recovery.
    let start_resp = start_result.with_context(|| {
        format!(
            "{name} migrated to a btrfs subvolume, but the restart request to the daemon \
             socket {} failed — run `hivectl start --agent {name}` to bring it back up",
            socket.display()
        )
    })?;
    if !start_resp.ok {
        bail!(
            "{name} migrated to a btrfs subvolume, but restarting it failed: {} — run \
             `hivectl start --agent {name}` to retry",
            start_resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    wait_for_dags(socket, start_resp.queued_dags.unwrap_or_default(), false)
        .await
        .with_context(|| {
            format!(
                "{name} migrated to a btrfs subvolume, but its restart job failed — run \
                 `hivectl start --agent {name}` to retry"
            )
        })?;
    println!("upgraded {name} to a btrfs subvolume and restarted it");
    Ok(())
}

/// `subvol snapshot create <agent> --label <label>` — create a read-only
/// btrfs snapshot of an agent's state subvolume. Unlike `upgrade`, this does
/// NOT stop the agent: btrfs snapshots are atomic + consistent to take
/// against a live subvolume. `label` is mandatory, must start with
/// `hive-`, and is otherwise restricted to `[A-Za-z0-9_-]` (no `.` at all
/// — mara: "we are making up the rules here, lets go strict"). hive-priv
/// enforces the same rules server-side, so this check is
/// belt-and-suspenders (fail fast client-side with a clear message).
async fn subvol_snapshot_create(name: &str, label: String) -> Result<()> {
    if !agent_exists(name)? {
        bail!("no agent named {name:?} (no state dir under the agents root)");
    }
    if !label.starts_with("hive-") {
        bail!("snapshot label {label:?} must start with \"hive-\"");
    }
    if !label
        .bytes()
        .all(|b| b.is_ascii_alphanumeric() || matches!(b, b'_' | b'-'))
    {
        bail!(
            "snapshot label {label:?} must be [A-Za-z0-9_-] only (no \".\" — hive-priv rejects it)"
        );
    }
    let path = hive_c0re::priv_client::snapshot_agent_subvolume(name, &label)
        .await
        .with_context(|| format!("snapshot {name} state subvolume (label {label:?})"))?;
    println!("{path}");
    Ok(())
}

/// `subvol snapshot delete <agent> <label>` — remove a snapshot created by
/// `subvol snapshot create`.
async fn subvol_snapshot_delete(name: &str, label: &str) -> Result<()> {
    hive_c0re::priv_client::delete_agent_snapshot(name, label)
        .await
        .with_context(|| format!("delete {name} snapshot (label {label:?})"))?;
    println!("deleted snapshot {label:?} for {name}");
    Ok(())
}

/// `subvol snapshot send <agent> <label> [--parent <label>] --dest <file>` —
/// export a snapshot to a local file via `btrfs send`. `dest` must be a
/// bare filename (hive-priv rejects anything else via the credential-name
/// charset check, which excludes `/`), written under the migrate-staging
/// dir. Prints the resulting file's host path.
async fn subvol_snapshot_send(
    name: &str,
    label: &str,
    parent: Option<&str>,
    dest: &str,
) -> Result<()> {
    let path = hive_c0re::priv_client::send_agent_snapshot_to_file(name, label, parent, dest)
        .await
        .with_context(|| format!("send {name} snapshot (label {label:?}) to file {dest:?}"))?;
    println!("{path}");
    Ok(())
}

/// Render a hive-wide stop/start response: one `<verb>: <name>` line per
/// touched container, then surface any aggregated per-target failure as a
/// non-zero exit. `verb` is the past-tense word printed per item
/// (`stopped` / `started`).
fn render_lifecycle(resp: &hive_host_sock::HostResponse, verb: &str) -> Result<()> {
    let items = resp.agents.as_deref().unwrap_or(&[]);
    if items.is_empty() {
        println!("{verb}: nothing matched the requested scope");
    } else {
        for item in items {
            println!("{verb}: {item}");
        }
    }
    if !resp.ok {
        bail!(
            "{verb}: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    Ok(())
}

/// Reject usernames containing `:` (field separator) or control chars
/// that would corrupt the htpasswd file format.
fn validate_htpasswd_username(username: &str) -> Result<()> {
    if username.is_empty() {
        bail!("username must not be empty");
    }
    if username.contains(':') {
        bail!("username must not contain ':' (htpasswd field separator)");
    }
    if username.chars().any(char::is_control) {
        bail!("username must not contain control characters");
    }
    Ok(())
}