A paused agent keeps its container, its claude session and its dashboard/todo servers up, but stops driving turns. Messages queue unacked and are drained on resume. The whole protocol is a single marker file, `<harness>/paused`. That directory is already a bind-mount shared between host and container, so both sides just stat the same path: the harness reads it to decide whether to drive a turn, hive-c0re reads it to render the badge and writes/removes it for `hivectl pause|resume`. No new wire protocol, no container round-trip, and it is sticky across restarts by construction. Not calling `recv_next` while paused *is* the queueing semantic, so there is no fencing to get wrong: reminders buffer in their unbounded channel, the todo `Notify` permit coalesces, and a `request_next_turn` that raced the pause survives because the gate sits above `self_continue.take()`. Graceful stop is handled host-side rather than in the harness: a paused agent provably has no turn in flight, so `run_signal` skips the fence entirely instead of eating the full `GRACEFUL_STOP_TIMEOUT` waiting for a checkpoint turn that will never run. `paused` is reported on `ContainerView` / `AgentStatusRow` for the dashboard, orthogonal to `running` and reported for stopped containers too. Closes: hyperhive/hyperhive issue 2271
205 lines
7.2 KiB
Rust
205 lines
7.2 KiB
Rust
//! `hivectl agents` — container lifecycle over the host admin socket
|
|
//! (list/restart/restart-all/spawn/kill/destroy/rebuild/set-parent).
|
|
|
|
use std::path::Path;
|
|
|
|
use anyhow::{Context as _, Result, bail};
|
|
use hive_host_sock::HostRequest;
|
|
|
|
use crate::cli::AgentsCmd;
|
|
use crate::dag_progress::wait_for_dags;
|
|
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_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 = 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::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:<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 = crate::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
|
|
}
|
|
|
|
/// `hivectl agents pause|resume` — flip the agent's pause marker. Not a
|
|
/// DAG, so there's nothing to wait on: the daemon writes the marker and
|
|
/// the harness picks it up on its next poll.
|
|
async fn set_paused(socket: &Path, name: &str, paused: bool) -> Result<()> {
|
|
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 {
|
|
let verb = if paused { "paused" } else { "resumed" };
|
|
println!("{verb}: {name}");
|
|
Ok(())
|
|
} else {
|
|
bail!(
|
|
"{} {name}: {}",
|
|
if paused { "pause" } else { "resume" },
|
|
resp.error.as_deref().unwrap_or("unknown error")
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Dispatch `hivectl agents <verb>` — container lifecycle over the host
|
|
/// admin socket.
|
|
pub(crate) async fn run_agents(socket: &Path, cmd: AgentsCmd) -> Result<()> {
|
|
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,
|
|
AgentsCmd::Pause { name } => set_paused(socket, &name, true).await,
|
|
AgentsCmd::Resume { name } => set_paused(socket, &name, false).await,
|
|
AgentsCmd::Spawn { name } => {
|
|
let name = crate::util::parse_ident(&name)?;
|
|
render(crate::client::request(socket, HostRequest::Spawn { name }).await?)
|
|
}
|
|
AgentsCmd::RequestSpawn { name } => {
|
|
let name = crate::util::parse_ident(&name)?;
|
|
render(crate::client::request(socket, HostRequest::RequestSpawn { name }).await?)
|
|
}
|
|
AgentsCmd::Kill { name } => {
|
|
let name = crate::util::parse_ident(&name)?;
|
|
render(crate::client::request(socket, HostRequest::Kill { name }).await?)
|
|
}
|
|
AgentsCmd::Destroy { name, purge } => {
|
|
let name = crate::util::parse_ident(&name)?;
|
|
render(crate::client::request(socket, HostRequest::Destroy { name, purge }).await?)
|
|
}
|
|
AgentsCmd::Rebuild { name } => {
|
|
let name = crate::util::parse_ident(&name)?;
|
|
render(crate::client::request(socket, HostRequest::Rebuild { name }).await?)
|
|
}
|
|
AgentsCmd::SetParent {
|
|
child,
|
|
parent,
|
|
root,
|
|
} => {
|
|
let child = crate::util::parse_ident(&child)?;
|
|
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?,
|
|
)
|
|
}
|
|
}
|
|
}
|