hyperhive/hivectl/src/agents.rs

//! `hivectl agent <name>` — everything scoped to ONE managed agent:
//! container lifecycle over the host admin socket (restart/pause/resume/
//! spawn/kill/destroy/rebuild/set-limits/choom/watch), plus the
//! `quota` and `subvol` groups, whose handlers live in their own modules.
//! `agents_list` (`hivectl list-agents`) is the one genuinely hive-wide
//! read that lives in this module too since it shares the same daemon
//! request as everything else here, even though it's dispatched from a
//! top-level `Cmd` variant, not `AgentCmd`.

use std::path::Path;

use anyhow::{Context as _, Result, bail};
use hive_host_sock::HostRequest;

use crate::cli::{AgentCmd, AgentQuotaCmd};
use crate::dag_progress::wait_for_nodes;
use crate::util::render;

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

/// `hivectl agent <name> start` — start an EXISTING agent container.
/// Fails immediately (no request even sent) if `name` has no state dir
/// at all, rather than silently resolving to an empty scope. Reuses the
/// exact hive-wide `hivectl start` DAG (`HostRequest::Start`'s
/// scope-based path), just scoped to this one name — a separate
/// per-agent wire request would have duplicated logic this scope-based
/// one already covers.
///
/// `paused`: if the agent is already running, sets the pause marker in
/// place and returns without submitting a start DAG at all (nothing to
/// start). Otherwise sets the marker *before* the start request, so the
/// container comes up paused rather than racing the harness's own
/// pause-gate poll against a start that's already in flight — the same
/// "pausing a stopped agent makes it come up paused" guarantee `pause`
/// already provides on its own.
async fn agents_start(socket: &Path, name: &str, paused: bool) -> Result<()> {
    if !crate::util::agent_exists(socket, name).await? {
        bail!(
            "no such agent: '{name}' (no state dir under {}/) — use 'hivectl agent {name} create' to provision a brand-new agent",
            hive_host_sock::AGENTS_ROOT
        );
    }
    if paused {
        let already_running = agent_running(socket, name).await?;
        // `no_wait = true`: correctness here doesn't come from waiting —
        // the pause DAG's `AgentWindow` brace and the `Start` request's
        // `SetWanted` head both declare `Resource::Agent`, and this call's
        // `insert_job` has already returned (so the pause DAG's claim is
        // registered) before `Start` is even sent, so the lease itself
        // orders the marker write ahead of the container boot. Waiting for
        // the *pause DAG* to finish
        // would be actively wrong when the agent isn't running yet: its
        // `PauseDrain` can't be acked until the harness is up to see the
        // marker, which is the very thing `Start` (below) is about to
        // cause — waiting here would just block for `PAUSE_ACK_TIMEOUT`
        // for an ack that can only land after this call returns.
        set_paused(socket, name, true, true).await?;
        if already_running {
            eprintln!("'{name}' is already running — paused in place, not (re)started");
            return Ok(());
        }
    }
    render(
        crate::client::request(
            socket,
            HostRequest::Start {
                scope: hive_host_sock::LifecycleScope {
                    agent_names: vec![name.to_owned()],
                    ..Default::default()
                },
            },
        )
        .await?,
    )
}

/// Whether `name`'s container is currently running, per the same
/// `AgentStatus` roster `hivectl list-agents` renders. No per-agent wire
/// request exists for this (nor should one, for a single boolean a rarely
/// called CLI flag needs) — filter the hive-wide roster down to one row.
async fn agent_running(socket: &Path, name: &str) -> Result<bool> {
    let resp = crate::client::request(socket, HostRequest::AgentStatus)
        .await
        .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !resp.ok {
        bail!(
            "agent status: {}",
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    Ok(resp
        .agent_statuses
        .unwrap_or_default()
        .iter()
        .any(|r| r.name == name && r.running))
}

/// `hivectl agent <name> stop` — graceful-only stop (signal → drain →
/// reconcile), never escalating to a hard kill. Reuses the hive-wide
/// `hivectl stop --graceful` DAG (`HostRequest::Stop`'s scope-based
/// path), scoped to this one name. Distinct from `AgentCmd::Kill`, which
/// hard-stops via a separate DAG that does escalate.
async fn agents_stop(socket: &Path, name: &str) -> Result<()> {
    render(
        crate::client::request(
            socket,
            HostRequest::Stop {
                scope: hive_host_sock::LifecycleScope {
                    agent_names: vec![name.to_owned()],
                    ..Default::default()
                },
                graceful: true,
            },
        )
        .await?,
    )
}

/// `hivectl list-agents` — 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.
pub(crate) async fn agents_list(socket: &Path, json: bool) -> Result<()> {
    let resp = crate::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::container::AgentStatusRow| -> String {
        let mut s = if r.running { "running" } else { "stopped" }.to_owned();
        if r.paused {
            s.push_str(" paused");
        }
        if r.needs_login {
            s.push_str(" needs-login");
        }
        if r.needs_update {
            s.push_str(" needs-update");
        }
        s
    };
    let headers = ["NAME", "STATUS", "REV", "REMIND"];
    let table: Vec<[String; 4]> = rows
        .iter()
        .map(|r| {
            [
                r.name.clone(),
                status_of(r),
                r.deployed_sha.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(())
}

/// `hivectl agent <name> pause|resume` — flip the agent's pause marker.
///
/// **Resume** is not a DAG — nothing to wait on: the daemon writes the
/// marker directly and the harness picks it up on its next poll.
///
/// **Pause** rides the job queue (`PauseSignal → PauseDrain`) instead —
/// see `handle_set_paused`'s doc comment on the daemon side for why. This
/// fn's `paused` branch prints once the DAG is *queued*, then optionally
/// waits for the harness's ack via `wait_for_nodes` (skippable with
/// `no_wait`, same knob `agents_restart` exposes).
///
/// Checks existence first, same as `agents_start` above — the daemon side
/// (`Coordinator::set_paused`) writes the marker file unconditionally via
/// the priv-helper and has no notion of "no such agent", so without this
/// check `hivectl agent typo-name resume` would exit 0 and print
/// `resumed: typo-name` for a name that was never a real agent.
async fn set_paused(socket: &Path, name: &str, paused: bool, no_wait: bool) -> Result<()> {
    if !crate::util::agent_exists(socket, name).await? {
        bail!(
            "no such agent: '{name}' (no state dir under {}/)",
            hive_host_sock::AGENTS_ROOT
        );
    }
    let resp = crate::client::request(
        socket,
        HostRequest::SetPaused {
            name: crate::util::parse_ident(name)?,
            paused,
        },
    )
    .await
    .with_context(|| format!("connect to daemon socket {}", socket.display()))?;
    if !resp.ok {
        bail!(
            "{} {name}: {}",
            if paused { "pause" } else { "resume" },
            resp.error.as_deref().unwrap_or("unknown error")
        );
    }
    if paused {
        println!("pause queued: {name}");
        wait_for_nodes(socket, resp.queued_dags.unwrap_or_default(), no_wait).await
    } else {
        println!("resumed: {name}");
        Ok(())
    }
}

/// Dispatch `hivectl agent <name> <verb>` — container lifecycle over the
/// host admin socket, all scoped to the single `name` hoisted from the
/// parent command.
pub(crate) async fn run_agent(socket: &Path, name: &str, cmd: AgentCmd) -> Result<()> {
    match cmd {
        AgentCmd::Restart { no_wait } => agents_restart(socket, name, no_wait).await,
        AgentCmd::Pause { no_wait } => set_paused(socket, name, true, no_wait).await,
        AgentCmd::Resume => set_paused(socket, name, false, false).await,
        AgentCmd::Start { paused } => agents_start(socket, name, paused).await,
        AgentCmd::Create => {
            let name = crate::util::parse_ident(name)?;
            render(crate::client::request(socket, HostRequest::Spawn { name }).await?)
        }
        AgentCmd::RequestCreate => {
            let name = crate::util::parse_ident(name)?;
            render(crate::client::request(socket, HostRequest::RequestSpawn { name }).await?)
        }
        AgentCmd::Stop => agents_stop(socket, name).await,
        AgentCmd::Kill => {
            let name = crate::util::parse_ident(name)?;
            render(crate::client::request(socket, HostRequest::Kill { name }).await?)
        }
        AgentCmd::Destroy { purge } => {
            let name = crate::util::parse_ident(name)?;
            render(crate::client::request(socket, HostRequest::Destroy { name, purge }).await?)
        }
        AgentCmd::Rebuild => {
            let name = crate::util::parse_ident(name)?;
            render(crate::client::request(socket, HostRequest::Rebuild { name }).await?)
        }
        AgentCmd::SetLimits {
            cpu_quota,
            memory_max,
            reset,
        } => {
            let name = crate::util::parse_ident(name)?;
            // `--reset` is the only way to reach an all-`None` request;
            // clap rejects a bare `set-limits` with neither flag, so a
            // forgotten value can't silently clear the overrides.
            let (cpu_quota, memory_max) = if reset {
                (None, None)
            } else {
                (cpu_quota, memory_max)
            };
            render(
                crate::client::request(
                    socket,
                    HostRequest::SetResourceLimits {
                        name,
                        cpu_quota,
                        memory_max,
                    },
                )
                .await?,
            )
        }
        AgentCmd::Choom { resume_session } => {
            crate::choom::choom(socket, name, resume_session.as_deref()).await
        }
        AgentCmd::Watch => crate::watch::watch(socket, name).await,
        // `quota` and `subvol` keep their own modules — these arms are
        // just the reparenting glue that hoists `name` in from the
        // parent `agent <name>` command.
        AgentCmd::Quota { cmd } => match cmd {
            AgentQuotaCmd::Show => crate::quota::quota_show(socket, name).await,
            AgentQuotaCmd::Set { size } => crate::quota::quota_limit(socket, name, &size).await,
        },
        AgentCmd::Subvol { cmd } => crate::subvol::dispatch_subvol(socket, name, cmd).await,
    }
}