//! `hivectl agent ` — everything scoped to ONE managed agent: //! container lifecycle over the host admin socket (restart/pause/resume/ //! spawn/kill/destroy/rebuild/set-parent/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 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 { 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 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", "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:>() .join(" ") .trim_end() .to_owned() }; let header_cells: Vec = headers.iter().map(|h| (*h).to_owned()).collect(); println!("{}", fmt_row(&header_cells)); for row in &table { println!("{}", fmt_row(row)); } Ok(()) } /// `hivectl agent 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 ` — 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::SetParent { parent, root } => { let child = crate::util::parse_ident(name)?; let new_parent = if root { None } else { parent.map(|p| crate::util::parse_ident(&p)).transpose()? }; render( crate::client::request(socket, HostRequest::SetParent { child, new_parent }) .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 ` 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, } }