1261b525d6
A swarm runs one homeserver and every hive on it logged in as the same `@hive:` localpart, holding the same access token out of one swarm-wide store path. That is one matrix identity for N hives: the homeserver cannot attribute an action to the hive that took it, and revoking one hive's standing revokes every hive's. Three changes, and the third is the one that makes the other two real: - **The localpart carries the hive's name** (`hive-<hive>`), derived in one place, `swarm_secret_client::matrix::hive_localpart`. `hive-matrix.nix` renders the same string as the appservice registration's `sender_localpart`, so the shared account stops being created rather than merely stops being used. - **The store path is templated by hive**, not a constant. The "a swarm runs one homeserver, so this is a constant rather than a parameter" rationale went with it; it stopped holding the moment two hives shared the homeserver it describes. - **The path moved out from under the grant every hive has.** It sat at `swarm/services/matrix/sender-token`, inside the `secret/data/swarm/services/*` read stanza `policy::render` gives every hive. It now sits under that hive's own stanza, `secret/data/swarm/hives/<hive>/*`, which interpolates the reader's name — so a hive reads its own token and is refused another's. The policy renderer itself is unchanged: narrowing the `services/*` grant would break the OIDC-secret read it exists for, and moving the credential is what this needed instead. A policy test walks the rendered stanzas and asserts none of hive alpha's covers hive beta's sender token, so a later stanza that widened it fails here. `swarm-matrix-ctl` takes a new required `MATRIX_MINT_HIVE` and writes that hive's path; its store grant in `swarm-bao.nix` follows, scoped to one hive's leaf via the new `deploy.bao.matrixCtlHiveName` (defaulting to this host's `hiveName`) rather than a `hives/*` wildcard, which would hand the matrix container every hive's token back. Migration: no outage at deploy. `ensure_hive_user` short-circuits on the local token file, so a hive keeps running on what it has; with no such file it reads the new per-hive path, finds nothing, and falls through to the existing register-or-appservice-login ladder against its own localpart — which needs only the per-hive `as_token` on local disk. The old shared object is read by nothing afterwards. Rooms do not follow the identity, and that is the one operator step; both ways out are written into `docs/integrations/matrix.md`. No admin standing is granted to the per-hive accounts: `admin_execute` stays empty and the assertion pinning it is untouched.
1176 lines
50 KiB
Rust
1176 lines
50 KiB
Rust
use std::path::Path;
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_host_sock::{HostRequest, HostResponse, LifecycleScope};
|
|
use hive_priv_sock::{InfraAction, InfraContainer};
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
|
|
use crate::actions;
|
|
use crate::coordinator::Coordinator;
|
|
use crate::lifecycle;
|
|
|
|
pub async fn serve(socket: &Path, coord: Arc<Coordinator>) -> Result<()> {
|
|
// Prefer a socket passed by systemd socket-activation (LISTEN_FDS).
|
|
// When running under a `.socket` unit, systemd has already created,
|
|
// bound, and chmod-ed the socket for us — we just accept on it.
|
|
// Fall back to the traditional bind path when not socket-activated
|
|
// (direct invocation, dev, tests).
|
|
let listener = {
|
|
let mut listenfd = listenfd::ListenFd::from_env();
|
|
if let Some(std_listener) = listenfd
|
|
.take_unix_listener(0)
|
|
.context("take socket-activated unix listener")?
|
|
{
|
|
std_listener.set_nonblocking(true)?;
|
|
UnixListener::from_std(std_listener)
|
|
.context("convert socket-activated listener to tokio")?
|
|
} else {
|
|
// Standalone: create parent dir, remove any stale socket, bind.
|
|
if let Some(parent) = socket.parent() {
|
|
std::fs::create_dir_all(parent)
|
|
.with_context(|| format!("create socket parent {}", parent.display()))?;
|
|
}
|
|
if socket.exists() {
|
|
std::fs::remove_file(socket).context("remove stale socket")?;
|
|
}
|
|
UnixListener::bind(socket)
|
|
.with_context(|| format!("bind admin socket {}", socket.display()))?
|
|
}
|
|
};
|
|
tracing::info!(socket = %socket.display(), hyperhive_flake = %coord.hyperhive_flake, "hive-c0re admin listening");
|
|
|
|
loop {
|
|
let (stream, _) = listener.accept().await.context("accept connection")?;
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = handle(stream, coord).await {
|
|
tracing::warn!(error = ?e, "connection failed");
|
|
}
|
|
});
|
|
}
|
|
}
|
|
|
|
async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
|
|
let (read, mut write) = stream.into_split();
|
|
let mut reader = BufReader::new(read);
|
|
let mut line = String::new();
|
|
|
|
loop {
|
|
line.clear();
|
|
let n = reader.read_line(&mut line).await?;
|
|
if n == 0 {
|
|
return Ok(());
|
|
}
|
|
let req = match serde_json::from_str::<HostRequest>(line.trim()) {
|
|
Ok(req) => req,
|
|
Err(e) => {
|
|
write_response(
|
|
&mut write,
|
|
&HostResponse::error(format!("parse error: {e}")),
|
|
)
|
|
.await?;
|
|
continue;
|
|
}
|
|
};
|
|
// The one verb that answers with more than one response — see its
|
|
// doc comment. Hands `write` over and stops reading further
|
|
// requests on this connection; a client wanting anything else
|
|
// opens a fresh one.
|
|
if matches!(req, HostRequest::SubscribeAgentStatus) {
|
|
return stream_agent_status(write, coord).await;
|
|
}
|
|
let resp = dispatch(&req, coord.clone()).await;
|
|
write_response(&mut write, &resp).await?;
|
|
}
|
|
}
|
|
|
|
/// Serialize one `HostResponse` as a JSON line and flush it. Shared by the
|
|
/// ordinary one-response-per-request path and `stream_agent_status`'s
|
|
/// multi-response one, so the two can't quietly drift on framing.
|
|
async fn write_response(
|
|
write: &mut tokio::net::unix::OwnedWriteHalf,
|
|
resp: &HostResponse,
|
|
) -> Result<()> {
|
|
let mut payload = serde_json::to_string(resp)?;
|
|
payload.push('\n');
|
|
write.write_all(payload.as_bytes()).await?;
|
|
write.flush().await?;
|
|
Ok(())
|
|
}
|
|
|
|
#[allow(
|
|
clippy::too_many_lines,
|
|
reason = "flat one-arm-per-HostRequest-variant router; each arm just \
|
|
delegates to a handler. Splitting the match would scatter the \
|
|
wire-command routing without shrinking it."
|
|
)]
|
|
async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
|
|
let result: anyhow::Result<HostResponse> = async {
|
|
Ok(match req {
|
|
HostRequest::Spawn { name } => handle_spawn(&coord, name.as_str()).await?,
|
|
HostRequest::RequestSpawn { name } => {
|
|
tracing::info!(%name, "request_spawn");
|
|
let id = coord.approvals.submit_kind(
|
|
name.as_str(),
|
|
hive_sh4re::approvals::ApprovalKind::Spawn,
|
|
"",
|
|
None,
|
|
"operator",
|
|
None,
|
|
)?;
|
|
tracing::info!(%id, %name, "spawn approval queued");
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::Kill { name } => submit_single(&coord, name.as_str(), Verb::Kill).await,
|
|
HostRequest::Restart { name } => {
|
|
submit_single(&coord, name.as_str(), Verb::Restart).await
|
|
}
|
|
HostRequest::SetPaused { name, paused } => {
|
|
handle_set_paused(&coord, name, *paused).await
|
|
}
|
|
HostRequest::RestartScoped { scope, graceful } => {
|
|
handle_restart_scoped(&coord, scope, *graceful).await?
|
|
}
|
|
HostRequest::Stop { scope, graceful } => {
|
|
// Resolve the scope to explicit container names at the entry
|
|
// point, then operate on names — never pass the bare "all
|
|
// agents" flag deeper (it'd force every consumer, incl. the
|
|
// graceful-stop queue, to re-expand it).
|
|
let agents = scoped_agents(scope).await?;
|
|
let infra = scoped_infra(scope);
|
|
// On a broad stop, remember which agents were actually
|
|
// running so a later broad `start` restores only those
|
|
// (not every configured container). A targeted `--agent`
|
|
// stop must not redefine the restore set.
|
|
if is_broad_scope(scope) {
|
|
let mut running = Vec::new();
|
|
for a in &agents {
|
|
if lifecycle::is_running(a).await {
|
|
running.push(a.clone());
|
|
}
|
|
}
|
|
coord.set_last_stopped_running(running);
|
|
}
|
|
handle_stop(&coord, &agents, &infra, *graceful).await?
|
|
}
|
|
HostRequest::Start { scope } => {
|
|
let mut agents = scoped_agents(scope).await?;
|
|
// A broad start restores only the set recorded at the
|
|
// last broad stop, if any. No record (cold "bring the
|
|
// hive up", or a daemon restart since the stop) → start
|
|
// all. Targeted `--agent` start is never filtered.
|
|
if is_broad_scope(scope)
|
|
&& let Some(prev) = coord.take_last_stopped_running()
|
|
{
|
|
agents.retain(|a| prev.contains(a));
|
|
}
|
|
let infra = scoped_infra(scope);
|
|
handle_start(&coord, &agents, &infra).await?
|
|
}
|
|
HostRequest::Destroy { name, purge } => {
|
|
actions::destroy(&coord, name.as_str(), *purge);
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::Rebuild { name } => {
|
|
submit_single(&coord, name.as_str(), Verb::Rebuild).await
|
|
}
|
|
HostRequest::QueueNodes { ids } => {
|
|
HostResponse::nodes(coord.job_queue.node_subtrees(ids))
|
|
}
|
|
HostRequest::List => HostResponse::list(lifecycle::list().await?),
|
|
// The agents root is ours and not world-traversable, so this
|
|
// question is only answerable on this side of the socket —
|
|
// see the request's doc comment for why the client asks.
|
|
HostRequest::AgentExists { name } => HostResponse::agent_exists(agent_exists(name)?),
|
|
HostRequest::AgentStatus => handle_agent_status(&coord).await,
|
|
// Intercepted in `handle()` before a request ever reaches
|
|
// `dispatch` — see `stream_agent_status`'s doc comment. This
|
|
// arm exists only so the match stays exhaustive; reaching it
|
|
// would mean a future caller invoked `dispatch` directly.
|
|
HostRequest::SubscribeAgentStatus => {
|
|
HostResponse::error("SubscribeAgentStatus is handled by the connection loop")
|
|
}
|
|
// The hive domain + per-surface public URLs are injected into
|
|
// c0re's service env by the hyperhive module; surface them so the
|
|
// operator CLI can fill in this hive's own identity (the
|
|
// federation peer-config block) and open the web surfaces
|
|
// (`hivectl open`).
|
|
HostRequest::Urls => HostResponse::urls(hive_urls()),
|
|
HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?),
|
|
HostRequest::Approve { id } => {
|
|
actions::approve(coord.clone(), *id).await?;
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::Deny { id } => {
|
|
actions::deny(&coord, *id, None)?;
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::SetParent { child, new_parent } => {
|
|
tracing::info!(%child, ?new_parent, "set_parent");
|
|
// Fire-and-forget, like every other queue-backed op:
|
|
// submit returns a DAG id immediately, the caller polls
|
|
// `QueueDag` (`hivectl`'s wait/progress loop) for the
|
|
// outcome instead of blocking here on the commit.
|
|
let inserted = coord.job_queue.insert_job(|b| {
|
|
vec![crate::job_queue::templates::reparent(
|
|
b,
|
|
vec![(child.clone(), new_parent.clone())],
|
|
)]
|
|
});
|
|
match inserted {
|
|
Ok(ids) => {
|
|
coord.emit_rebuild_queue_snapshot();
|
|
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
|
|
}
|
|
Err(e) => HostResponse::error(format!("queue reparent: {e}")),
|
|
}
|
|
}
|
|
HostRequest::SetResourceLimits {
|
|
name,
|
|
cpu_quota,
|
|
memory_max,
|
|
} => {
|
|
handle_set_resource_limits(
|
|
&coord,
|
|
name,
|
|
cpu_quota.as_deref(),
|
|
memory_max.as_deref(),
|
|
)
|
|
.await?
|
|
}
|
|
HostRequest::MatrixCreateUser { name, password } => {
|
|
handle_matrix_create_user(name, password.as_deref()).await?
|
|
}
|
|
HostRequest::MatrixSyncAdmin => handle_matrix_sync_admin().await?,
|
|
HostRequest::MatrixPromoteUser { name } => {
|
|
handle_matrix_promote_user(name.as_str()).await?
|
|
}
|
|
HostRequest::MatrixResetPassword { name } => {
|
|
handle_matrix_reset_password(name.as_str()).await?
|
|
}
|
|
HostRequest::MatrixInvite { user, room } => {
|
|
handle_matrix_invite(user, room.as_deref()).await?
|
|
}
|
|
HostRequest::ForgeCreateUser { name, password } => {
|
|
handle_forge_create_user(name, password.as_deref()).await?
|
|
}
|
|
HostRequest::ReconcileConfigStatus { agent, verbose } => {
|
|
crate::forge::reconcile_config_status(agent.as_str(), *verbose).await?
|
|
}
|
|
HostRequest::ReconcileConfigApply { agent, direction } => {
|
|
crate::forge::reconcile_config_apply(agent.as_str(), *direction).await?
|
|
}
|
|
HostRequest::GatewayCreateUser { username, password } => {
|
|
HostResponse::messages(vec![crate::gateway_nginx::create_user(username, password)?])
|
|
}
|
|
HostRequest::GatewayDeleteUser { username } => {
|
|
HostResponse::messages(vec![crate::gateway_nginx::delete_user(username)?])
|
|
}
|
|
HostRequest::GatewayListUsers => {
|
|
HostResponse::messages(crate::gateway_nginx::list_users()?)
|
|
}
|
|
HostRequest::SetAgentGithubToken { agent, token } => {
|
|
handle_set_agent_github_token(agent.as_str(), token).await?
|
|
}
|
|
HostRequest::QuotaEnable => handle_quota_enable().await?,
|
|
HostRequest::QuotaLimit { name, limit } => {
|
|
handle_quota_limit(name.as_str(), *limit).await?
|
|
}
|
|
HostRequest::QuotaShow { name } => {
|
|
handle_quota_show(name.as_ref().map(hive_types::Ident::as_str)).await?
|
|
}
|
|
HostRequest::UpgradeSubvolume { name } => {
|
|
handle_upgrade_subvolume(name.as_str()).await?
|
|
}
|
|
HostRequest::SnapshotSubvolume { name, label } => {
|
|
handle_snapshot_subvolume(name.as_str(), label).await?
|
|
}
|
|
HostRequest::DeleteSnapshot { name, label } => {
|
|
handle_delete_snapshot(name.as_str(), label).await?
|
|
}
|
|
HostRequest::SendSnapshot {
|
|
name,
|
|
label,
|
|
parent,
|
|
dest,
|
|
} => handle_send_snapshot(name.as_str(), label, parent.as_deref(), dest).await?,
|
|
HostRequest::PushSnapshot {
|
|
name,
|
|
label,
|
|
parent,
|
|
} => handle_push_snapshot(name.as_str(), label, parent.as_deref()).await?,
|
|
})
|
|
}
|
|
.await;
|
|
match result {
|
|
Ok(r) => r,
|
|
Err(e) => HostResponse::error(format!("{e:#}")),
|
|
}
|
|
}
|
|
|
|
/// Create + start the container for `name`, rolling back socket
|
|
/// registration and notifying the manager on failure.
|
|
async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
|
|
tracing::info!(%name, "spawn");
|
|
let agent_dir = crate::paths::agent_runtime_dir(name);
|
|
let hive = coord.hive_env();
|
|
let paths = Coordinator::agent_paths(name, agent_dir);
|
|
// lifecycle::spawn creates the runtime dir internally before start.
|
|
// MCP listener registration is event-driven: bind immediately on
|
|
// success so the harness can connect on its first turn without
|
|
// waiting for any poll interval.
|
|
match lifecycle::spawn(name, &hive, &paths).await {
|
|
Ok(()) => {
|
|
if let Err(e) = coord.power.set(name, crate::power::Wanted::Up) {
|
|
tracing::warn!(%name, error = ?e, "agent_power: set wanted=up failed");
|
|
}
|
|
// Bind the MCP listener now that the container is starting up.
|
|
// The harness connects to this socket on its first turn.
|
|
coord.register_agent(name)?;
|
|
crate::swarm_notices::notify(
|
|
"core",
|
|
Some(format!("spawned:{name}")),
|
|
format!("agent '{name}' spawned"),
|
|
None,
|
|
)
|
|
.await;
|
|
// Update tmpfiles.d so the new agent's dirs survive a reboot.
|
|
tokio::spawn(lifecycle::sync_tmpfiles());
|
|
}
|
|
Err(e) => {
|
|
// Spawn failed: register_agent was never called, so there is
|
|
// nothing to unregister. Notify the swarm and propagate.
|
|
crate::swarm_notices::notify(
|
|
"core",
|
|
Some(format!("spawned:{name}")),
|
|
format!("agent '{name}' spawn FAILED: {e:#}"),
|
|
None,
|
|
)
|
|
.await;
|
|
return Err(e);
|
|
}
|
|
}
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
/// `hivectl pause|resume` / the dashboard toggle: write or remove the
|
|
/// agent's pause marker.
|
|
///
|
|
/// **Resume** stays synchronous, direct marker removal — there's
|
|
/// nothing to acknowledge (a resumed agent just starts driving turns
|
|
/// again on its own next poll, no handshake needed), and nobody has
|
|
/// asked resume to wait.
|
|
///
|
|
/// **Pause** rides the job queue (`PauseSignal → PauseDrain`, see
|
|
/// `job_queue::power::pause_many`) instead of writing the marker
|
|
/// synchronously here: pausing wants the same "confirmed, not just
|
|
/// requested" signal a graceful stop gets from its `Signal → Drain`
|
|
/// pair, and the DAG's agent lease is what stops a pause from racing
|
|
/// an in-flight rebuild/stop of the same agent — a synchronous write
|
|
/// had no such protection. Still works on a stopped agent (the queued
|
|
/// `PauseSignal` writes the same sticky marker either way, container
|
|
/// running or not).
|
|
async fn handle_set_paused(
|
|
coord: &std::sync::Arc<Coordinator>,
|
|
name: &hive_types::Ident,
|
|
paused: bool,
|
|
) -> HostResponse {
|
|
if !paused {
|
|
if let Err(e) = Coordinator::set_paused(name, false).await {
|
|
return HostResponse::error(format!("set paused=false for {name}: {e}"));
|
|
}
|
|
tracing::info!(%name, "agent pause marker cleared");
|
|
// Refresh the dashboard's view so the paused badge flips without
|
|
// waiting for the next periodic rescan.
|
|
coord.rescan_containers_and_emit().await;
|
|
return HostResponse::success();
|
|
}
|
|
match crate::job_queue::power::pause_many(coord, std::slice::from_ref(&name.to_string())) {
|
|
Ok(ids) => {
|
|
tracing::info!(%name, "agent pause queued");
|
|
HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect())
|
|
}
|
|
Err(e) => HostResponse::error(format!("queue pause for {name}: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Collect per-agent status rows for `hivectl status` and the dashboard.
|
|
async fn handle_agent_status(coord: &Arc<Coordinator>) -> HostResponse {
|
|
let rows = crate::container_view::build_all(&coord.hive_env())
|
|
.await
|
|
.into_iter()
|
|
.map(hive_sh4re::container::AgentStatusRow::from)
|
|
.collect();
|
|
HostResponse::agent_statuses(rows)
|
|
}
|
|
|
|
/// `SubscribeAgentStatus` — ack once, then push one single-row
|
|
/// [`HostResponse::agent_statuses`] per agent-status change for as long as
|
|
/// the client stays connected. Consumes `write` (rather than borrowing it,
|
|
/// like every other handler here) because this is the one verb that keeps
|
|
/// writing after its own response, so `handle()` hands over ownership and
|
|
/// stops reading further requests on this connection once it calls this.
|
|
///
|
|
/// Rides the same `Coordinator.dashboard_events` broadcast channel the
|
|
/// dashboard's own SSE route reads (`dashboard/state_snapshot.rs`) —
|
|
/// `SetStatus` already triggers a `rescan_containers_and_emit` on every
|
|
/// status change (`socket_server/mod.rs::handle_set_status`), so that
|
|
/// channel already carries every event this needs; no separate plumbing.
|
|
async fn stream_agent_status(
|
|
mut write: tokio::net::unix::OwnedWriteHalf,
|
|
coord: Arc<Coordinator>,
|
|
) -> Result<()> {
|
|
let mut events = coord.dashboard_subscribe();
|
|
write_response(&mut write, &HostResponse::success()).await?;
|
|
loop {
|
|
match events.recv().await {
|
|
Ok(crate::dashboard_events::DashboardEvent::ContainerStateChanged {
|
|
container,
|
|
..
|
|
}) => {
|
|
let row = hive_sh4re::container::AgentStatusRow::from(container);
|
|
write_response(&mut write, &HostResponse::agent_statuses(vec![row])).await?;
|
|
}
|
|
// Every other event kind on this channel is dashboard-only
|
|
// (approvals, broker traffic, the queue, …) — nothing this
|
|
// subscriber asked for.
|
|
Ok(_) => {}
|
|
// Best-effort, same contract the dashboard's own live channel
|
|
// already has (see `HostRequest::SubscribeAgentStatus`'s doc
|
|
// comment) — a slow reader drops updates rather than stalling
|
|
// the broadcaster for everyone else on the channel.
|
|
Err(tokio::sync::broadcast::error::RecvError::Lagged(skipped)) => {
|
|
tracing::warn!(skipped, "agent-status subscribe: receiver lagged");
|
|
}
|
|
Err(tokio::sync::broadcast::error::RecvError::Closed) => return Ok(()),
|
|
}
|
|
}
|
|
}
|
|
|
|
// ---------------------------------------------------------------------------
|
|
// Matrix provisioning handlers
|
|
//
|
|
// The `hivectl matrix` subcommands used to run these in-process, which forced
|
|
// the standalone CLI to link the whole daemon crate (matrix-sdk, reqwest, …).
|
|
// They now run daemon-side over the host socket: the daemon already holds the
|
|
// register + sender tokens and the matrix creds dir. Each op returns the
|
|
// operator-facing lines hivectl used to `println!` in `HostResponse::messages`
|
|
// for the client to print verbatim.
|
|
// ---------------------------------------------------------------------------
|
|
|
|
/// Shared reqwest client for the matrix admin HTTP calls (30s timeout,
|
|
/// mirroring the old in-CLI client).
|
|
fn matrix_http_client() -> Result<reqwest::Client> {
|
|
reqwest::Client::builder()
|
|
.timeout(std::time::Duration::from_secs(30))
|
|
.build()
|
|
.context("build reqwest client")
|
|
}
|
|
|
|
/// True when `name` has a state dir under the agents root, i.e. it's a
|
|
/// managed agent rather than a bare (operator/human) account.
|
|
///
|
|
/// Used by the provisioning handlers below to tell an agent account from
|
|
/// a human one, and exposed over the socket as
|
|
/// [`HostRequest::AgentExists`] for clients that can't read the agents
|
|
/// root themselves (it's `0700` and owned by the daemon's user).
|
|
fn agent_exists(name: &hive_types::Ident) -> Result<bool> {
|
|
crate::paths::agent_state_dir(name)
|
|
.try_exists()
|
|
.with_context(|| format!("check agent state dir for {name}"))
|
|
}
|
|
|
|
/// Validate + persist an agent's CPU/memory overrides, then re-apply the
|
|
/// drop-in so the change lands without waiting for a rebuild.
|
|
///
|
|
/// Validation is here rather than only in `hivectl` because the values
|
|
/// are written verbatim into the systemd drop-in: a malformed
|
|
/// `CPUQuota=` makes systemd reject the unit, and the container stops
|
|
/// starting. Every client (CLI, dashboard, anything later) goes through
|
|
/// this path, so the guard belongs on this side of the socket.
|
|
///
|
|
/// `None`/`None` removes the agent's entry, returning it to the
|
|
/// hive-wide defaults.
|
|
async fn handle_set_resource_limits(
|
|
coord: &Arc<Coordinator>,
|
|
name: &hive_types::Ident,
|
|
cpu_quota: Option<&str>,
|
|
memory_max: Option<&str>,
|
|
) -> Result<HostResponse> {
|
|
if let Some(value) = cpu_quota {
|
|
crate::resource_limits::validate_cpu_quota(value).map_err(anyhow::Error::msg)?;
|
|
}
|
|
// `validate_memory_max` returns the value to store, not just an
|
|
// ok/err verdict: a friendly spelling like "8GB" is accepted but
|
|
// normalized to the "8G" systemd's own parser actually takes.
|
|
let memory_max = memory_max
|
|
.map(crate::resource_limits::validate_memory_max)
|
|
.transpose()
|
|
.map_err(anyhow::Error::msg)?;
|
|
tracing::info!(%name, ?cpu_quota, ?memory_max, "set_resource_limits");
|
|
let limits = crate::resource_limits::AgentLimits {
|
|
cpu_quota: cpu_quota.map(ToOwned::to_owned),
|
|
memory_max,
|
|
};
|
|
// Goes through `meta::commit_resource_limits`, not the bare
|
|
// `resource_limits::set_limits`: the write has to be staged +
|
|
// committed under `META_LOCK` or it leaves the meta working tree
|
|
// dirty for the next `prepare_deploy` / `sync_agents` to trip over.
|
|
crate::meta::commit_resource_limits(name.as_str(), &limits).await?;
|
|
|
|
// Re-apply the drop-in straight away — same three lines as the job
|
|
// queue's `WriteDropin` node. Without this the new values would sit
|
|
// in the JSON until the agent's next spawn or rebuild.
|
|
let agent_dir = crate::paths::agent_runtime_dir(name.as_str());
|
|
let hive = coord.hive_env();
|
|
let paths = Coordinator::agent_paths(name.as_str(), agent_dir);
|
|
crate::lifecycle::write_dropins(name.as_str(), &hive, &paths).await?;
|
|
|
|
let (cpu, mem) = crate::resource_limits::effective(
|
|
name.as_str(),
|
|
&hive.agent_cpu_quota,
|
|
&hive.agent_memory_max,
|
|
);
|
|
Ok(HostResponse::messages(vec![format!(
|
|
"{name}: CPUQuota={cpu} MemoryMax={mem} — the cgroup cap itself is live now (restart \
|
|
the container if it's running and needs the new cap immediately), but the derived \
|
|
Claude/JSC heap ceiling is baked in at build time, so it needs a REBUILD \
|
|
(`hivectl agent {name} rebuild`) to actually track this change"
|
|
)]))
|
|
}
|
|
|
|
/// Guard: matrix provisioning needs a homeserver to provision against.
|
|
///
|
|
/// Answers "is one configured", not "is one running here" — the message names
|
|
/// both ways to get there, since a hive that talks to someone else's
|
|
/// homeserver never enables the local container at all.
|
|
fn require_matrix_present() -> Result<()> {
|
|
if crate::matrix::is_present() {
|
|
return Ok(());
|
|
}
|
|
anyhow::bail!(
|
|
"no matrix homeserver configured — set services.hyperhive.deploy.matrix.enable = true to run one \
|
|
here, or services.hyperhive.swarm.matrix.apiUrl to point at an existing one, before \
|
|
provisioning matrix users"
|
|
)
|
|
}
|
|
|
|
async fn handle_matrix_create_user(
|
|
name: &hive_types::Ident,
|
|
password: Option<&str>,
|
|
) -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let as_token =
|
|
crate::matrix::read_appservice_token().context("read matrix appservice token")?;
|
|
let client = matrix_http_client()?;
|
|
let mut out = Vec::new();
|
|
if agent_exists(name)? {
|
|
if password.is_some() {
|
|
// Agents auth by access_token, never by password — the
|
|
// boot-sweep provisioning path doesn't accept one. Refuse
|
|
// rather than silently dropping it.
|
|
anyhow::bail!(
|
|
"matrix create-user: a password is for non-agent (operator) accounts only; '{name}' is an agent which authenticates via access_token"
|
|
);
|
|
}
|
|
crate::matrix::ensure_user_for(&client, name.as_str(), &as_token)
|
|
.await
|
|
.with_context(|| format!("matrix create-user {name}"))?;
|
|
let path = Coordinator::agent_notes_dir(name).join("matrix-token");
|
|
out.push(format!("matrix: provisioned agent user '{name}'"));
|
|
out.push(format!("token persisted at: {}", path.display()));
|
|
} else {
|
|
let effective_password = match password {
|
|
Some(p) => p.to_owned(),
|
|
None => crate::matrix::random_password().context("generate random matrix password")?,
|
|
};
|
|
let token = crate::matrix::provision_user_token(
|
|
&client,
|
|
name.as_str(),
|
|
&as_token,
|
|
&effective_password,
|
|
)
|
|
.await
|
|
.with_context(|| format!("matrix create-user {name}"))?;
|
|
out.push(format!(
|
|
"matrix: provisioned user '{name}' (not an agent — token not persisted)"
|
|
));
|
|
out.push(format!("token: {token}"));
|
|
if password.is_some() {
|
|
out.push(
|
|
"password: set as supplied — use it to log into a matrix web client".to_owned(),
|
|
);
|
|
} else {
|
|
out.push(
|
|
"password: random throwaway (not surfaced — pass --password or --password-stdin to set one you can use)".to_owned(),
|
|
);
|
|
}
|
|
}
|
|
Ok(HostResponse::messages(out))
|
|
}
|
|
|
|
async fn handle_forge_create_user(
|
|
name: &hive_types::Ident,
|
|
password: Option<&str>,
|
|
) -> Result<HostResponse> {
|
|
if !crate::forge::is_present().await {
|
|
anyhow::bail!(
|
|
"hive-forge container not running — wait for hive-c0re to start it before provisioning forge users"
|
|
);
|
|
}
|
|
let mut out = Vec::new();
|
|
if agent_exists(name)? {
|
|
if password.is_some() {
|
|
// Agents authenticate by API token, never by password — refuse
|
|
// rather than silently dropping a supplied one.
|
|
anyhow::bail!(
|
|
"forge create-user: a password is for non-agent (operator) accounts only; '{name}' is an agent which authenticates via API token"
|
|
);
|
|
}
|
|
crate::forge::ensure_user_for(name.as_str())
|
|
.await
|
|
.with_context(|| format!("forge create-user {name}"))?;
|
|
let path = Coordinator::agent_notes_dir(name).join("forge-token");
|
|
out.push(format!("forge: provisioned agent user '{name}'"));
|
|
out.push(format!("token persisted at: {}", path.display()));
|
|
} else {
|
|
let token = crate::forge::provision_user_token(name.as_str(), password)
|
|
.await
|
|
.with_context(|| format!("forge create-user {name}"))?;
|
|
out.push(format!(
|
|
"forge: provisioned user '{name}' (not an agent — token not persisted)"
|
|
));
|
|
out.push(format!("token: {token}"));
|
|
if password.is_some() {
|
|
out.push("password: set as supplied — use it to log into the forge web UI".to_owned());
|
|
} else {
|
|
out.push(
|
|
"password: random throwaway (not surfaced — pass --password or --password-stdin to set one you can use)".to_owned(),
|
|
);
|
|
}
|
|
}
|
|
Ok(HostResponse::messages(out))
|
|
}
|
|
|
|
async fn handle_set_agent_github_token(agent: &str, token: &str) -> Result<HostResponse> {
|
|
crate::priv_client::write_agent_github_token(agent, token)
|
|
.await
|
|
.with_context(|| format!("write github-token for agent {agent}"))?;
|
|
Ok(HostResponse::messages(vec![format!(
|
|
"wrote github-token for agent '{agent}' \
|
|
(read live by the gh wrapper / git credential helper — no rebuild needed)"
|
|
)]))
|
|
}
|
|
|
|
async fn handle_quota_enable() -> Result<HostResponse> {
|
|
crate::priv_client::ensure_btrfs_quota()
|
|
.await
|
|
.context("enable btrfs qgroup accounting")?;
|
|
Ok(HostResponse::messages(vec![
|
|
"btrfs qgroup accounting enabled on the agent-state filesystem.".to_owned(),
|
|
"(usage may read 0 until btrfs finishes its background rescan)".to_owned(),
|
|
]))
|
|
}
|
|
|
|
async fn handle_quota_limit(name: &str, limit: Option<u64>) -> Result<HostResponse> {
|
|
crate::priv_client::set_subvolume_quota(name, limit)
|
|
.await
|
|
.with_context(|| format!("set quota for {name}"))?;
|
|
// Bare success: the client prints the human-readable confirmation from
|
|
// the value it sent (it holds the `human_bytes` formatter).
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_quota_show(name: Option<&str>) -> Result<HostResponse> {
|
|
let agents: Vec<String> = match name {
|
|
Some(n) => vec![n.to_owned()],
|
|
None => Coordinator::kept_state_names()
|
|
.into_iter()
|
|
.map(hive_types::Ident::into_string)
|
|
.collect(),
|
|
};
|
|
let mut rows = Vec::with_capacity(agents.len());
|
|
for agent in &agents {
|
|
match crate::priv_client::read_subvolume_usage(agent).await {
|
|
Ok((referenced, exclusive)) => rows.push(hive_host_sock::QuotaRow {
|
|
agent: agent.clone(),
|
|
referenced: Some(referenced),
|
|
exclusive: Some(exclusive),
|
|
note: None,
|
|
}),
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
// btrfs-progs prints "ERROR: ... quota not enabled" to stderr
|
|
// when qgroups are off; short-circuit the whole sweep with the
|
|
// enable hint (case-insensitive fragment match — the wording
|
|
// varies across btrfs-progs versions).
|
|
if msg.to_ascii_lowercase().contains("quota not enabled") {
|
|
return Ok(HostResponse::error(
|
|
"btrfs quota not enabled — run `hivectl quota-enable` first",
|
|
));
|
|
}
|
|
// A plain-dir agent (no subvolume) has no qgroup; note it
|
|
// inline and keep going rather than aborting the whole sweep.
|
|
rows.push(hive_host_sock::QuotaRow {
|
|
agent: agent.clone(),
|
|
referenced: None,
|
|
exclusive: None,
|
|
note: Some(format!("no qgroup data — plain dir or: {msg}")),
|
|
});
|
|
}
|
|
}
|
|
}
|
|
Ok(HostResponse::quota(rows))
|
|
}
|
|
|
|
async fn handle_upgrade_subvolume(name: &str) -> Result<HostResponse> {
|
|
crate::priv_client::upgrade_agent_subvolume(name)
|
|
.await
|
|
.with_context(|| format!("upgrade {name} state subvolume"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_snapshot_subvolume(name: &str, label: &str) -> Result<HostResponse> {
|
|
let path = crate::priv_client::snapshot_agent_subvolume(name, label)
|
|
.await
|
|
.with_context(|| format!("snapshot {name} state subvolume (label {label:?})"))?;
|
|
Ok(HostResponse::messages(vec![path]))
|
|
}
|
|
|
|
async fn handle_delete_snapshot(name: &str, label: &str) -> Result<HostResponse> {
|
|
crate::priv_client::delete_agent_snapshot(name, label)
|
|
.await
|
|
.with_context(|| format!("delete {name} snapshot (label {label:?})"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_send_snapshot(
|
|
name: &str,
|
|
label: &str,
|
|
parent: Option<&str>,
|
|
dest: &str,
|
|
) -> Result<HostResponse> {
|
|
let path = crate::priv_client::send_agent_snapshot_to_file(name, label, parent, dest)
|
|
.await
|
|
.with_context(|| format!("send {name} snapshot (label {label:?}) to file {dest:?}"))?;
|
|
Ok(HostResponse::messages(vec![path]))
|
|
}
|
|
|
|
/// Push a snapshot to the swarm's store. The network sibling of
|
|
/// [`handle_send_snapshot`]: nothing lands on this host, so success is
|
|
/// bare rather than a path.
|
|
async fn handle_push_snapshot(
|
|
name: &str,
|
|
label: &str,
|
|
parent: Option<&str>,
|
|
) -> Result<HostResponse> {
|
|
crate::snapshot_push::push_agent_snapshot(name, label, parent)
|
|
.await
|
|
.with_context(|| format!("push {name} snapshot (label {label:?}) to the swarm store"))?;
|
|
Ok(HostResponse::success())
|
|
}
|
|
|
|
async fn handle_matrix_sync_admin() -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let as_token =
|
|
crate::matrix::read_appservice_token().context("read matrix appservice token")?;
|
|
let client = matrix_http_client()?;
|
|
crate::matrix::ensure_hive_user(&client, &as_token)
|
|
.await
|
|
.context("matrix sync-admin")?;
|
|
let path = crate::matrix::sender_token_path();
|
|
Ok(HostResponse::messages(vec![
|
|
format!(
|
|
"matrix: the @{}: user is provisioned",
|
|
crate::matrix::hive_localpart()?
|
|
),
|
|
format!("token persisted at: {}", path.display()),
|
|
]))
|
|
}
|
|
|
|
async fn handle_matrix_promote_user(name: &str) -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let sender_token = crate::matrix::read_sender_token()?;
|
|
let client = matrix_http_client()?;
|
|
let server_name = crate::matrix::discover_server_name(&client)
|
|
.await
|
|
.context("discover matrix server_name")?;
|
|
crate::matrix::promote_user_to_admin(&client, &sender_token, name, &server_name)
|
|
.await
|
|
.with_context(|| format!("matrix promote-user {name}"))?;
|
|
Ok(HostResponse::messages(vec![format!(
|
|
"matrix: promoted @{name}:{server_name} to admin"
|
|
)]))
|
|
}
|
|
|
|
async fn handle_matrix_invite(user: &str, room: Option<&str>) -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let sender_token = crate::matrix::read_sender_token()?;
|
|
let client = matrix_http_client()?;
|
|
let server_name = crate::matrix::discover_server_name(&client)
|
|
.await
|
|
.context("discover matrix server_name")?;
|
|
let room_id = crate::matrix::invite_user(&client, &sender_token, user, room, &server_name)
|
|
.await
|
|
.with_context(|| format!("matrix invite {user}"))?;
|
|
let target = if user.starts_with('@') {
|
|
user.to_owned()
|
|
} else {
|
|
format!("@{user}:{server_name}")
|
|
};
|
|
Ok(HostResponse::messages(vec![format!(
|
|
"matrix: invited {target} to {room_id}"
|
|
)]))
|
|
}
|
|
|
|
async fn handle_matrix_reset_password(name: &str) -> Result<HostResponse> {
|
|
require_matrix_present()?;
|
|
let sender_token = crate::matrix::read_sender_token()?;
|
|
let client = matrix_http_client()?;
|
|
let server_name = crate::matrix::discover_server_name(&client)
|
|
.await
|
|
.context("discover matrix server_name")?;
|
|
crate::matrix::reset_user_password(&client, &sender_token, name, &server_name)
|
|
.await
|
|
.with_context(|| format!("matrix reset-password {name}"))?;
|
|
// Password is persisted by reset_user_password.
|
|
let pw_path = crate::paths::matrix_creds_dir().join(format!("{name}-password"));
|
|
Ok(HostResponse::messages(vec![
|
|
format!("matrix: password for @{name}:{server_name} reset"),
|
|
format!("password persisted at: {}", pw_path.display()),
|
|
format!("next: hivectl matrix create-user {name} # mints a fresh access token"),
|
|
]))
|
|
}
|
|
|
|
/// Single-agent queue verbs the admin socket exposes. Each submits the
|
|
/// matching DAG (persisting the `wanted` intent, serializing on the
|
|
/// agent's lease, with the transient/crash-watch suppression the old
|
|
/// direct lifecycle calls lacked) and returns the DAG id for the
|
|
/// client's wait loop.
|
|
#[derive(Clone, Copy)]
|
|
enum Verb {
|
|
/// Stop DAG (`wanted = Offline`; Reconcile kills + unregisters +
|
|
/// fires `Killed`).
|
|
Kill,
|
|
/// Restart DAG (`wanted = Up`; mechanical stop + reconcile-start).
|
|
Restart,
|
|
/// Rebuild DAG — the Swap tail owns the manager `Rebuilt` events +
|
|
/// kick, so the CLI path can't drift from the dashboard's.
|
|
Rebuild,
|
|
}
|
|
|
|
async fn submit_single(coord: &Arc<Coordinator>, name: &str, verb: Verb) -> HostResponse {
|
|
use crate::job_queue::power;
|
|
// A single-target op is the N-target one with N = 1 — the shapes are
|
|
// identical, so there is no separate builder to keep in sync.
|
|
let targets = [name.to_owned()];
|
|
let ids = match verb {
|
|
Verb::Kill => {
|
|
tracing::info!(%name, "kill");
|
|
power::stop_many(coord, &targets, false).await
|
|
}
|
|
Verb::Restart => {
|
|
tracing::info!(%name, "restart");
|
|
power::restart_many(coord, &targets, false).await
|
|
}
|
|
Verb::Rebuild => {
|
|
tracing::info!(%name, "rebuild");
|
|
// Not a power op: a rebuild's shape doesn't depend on live state,
|
|
// so it is a plain template insert rather than a `*_many` gather.
|
|
let inserted = coord
|
|
.job_queue
|
|
.insert_job(|b| crate::job_queue::templates::rebuild(b, name, true));
|
|
if inserted.is_ok() {
|
|
coord.emit_rebuild_queue_snapshot();
|
|
}
|
|
inserted
|
|
}
|
|
};
|
|
match ids {
|
|
Ok(ids) => HostResponse::queued(ids.into_iter().map(hive_jobq::NodeId::get).collect()),
|
|
Err(e) => HostResponse::error(format!("queue insert failed: {e}")),
|
|
}
|
|
}
|
|
|
|
/// Stop the given `agents` (resolved logical names) then `infra` containers
|
|
/// (`hivectl stop`). Agents go down before infra so they're not mid-request
|
|
/// against a forge/matrix that's already gone. Per-target failures are
|
|
/// aggregated rather than aborting on the first error, mirroring
|
|
/// `finish_lifecycle` below. Callers resolve the [`LifecycleScope`] to these
|
|
/// explicit name lists up front — this never sees the "all" flag.
|
|
///
|
|
/// Every agent rides the job queue: a `graceful` stop submits the
|
|
/// quiesce DAG (signal → drain → reconcile-stop; all drains overlap),
|
|
/// a hard stop a plain stop DAG — both persist `wanted = Offline` and
|
|
/// serialize on the agent's lease so nothing races an in-flight
|
|
/// rebuild. The response carries the DAG ids so `hivectl` can wait
|
|
/// with per-node progress. Infra containers have no harness / lease
|
|
/// and stay direct + synchronous.
|
|
async fn handle_stop(
|
|
coord: &Arc<Coordinator>,
|
|
agents: &[String],
|
|
infra: &[InfraContainer],
|
|
graceful: bool,
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?agents, ?infra, graceful, "stop");
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
|
|
// One insert for all targeted agents — a per-agent stop subgraph each
|
|
// (`SetWanted(Offline) → [Signal → Drain →] Reconcile`), independent
|
|
// roots that run concurrently on their own leases.
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::stop_many(coord, agents, graceful).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue stop: {e}")),
|
|
}
|
|
}
|
|
|
|
// Agents go down before infra so they're not mid-request against a
|
|
// forge/matrix that's already gone. Hard stops are quick kills —
|
|
// await their DAGs (bounded) before touching infra. Graceful stops
|
|
// keep the immediate return (drains take minutes and the
|
|
// agents-then-infra race pre-existed there).
|
|
if !graceful && !infra.is_empty() {
|
|
await_dags(coord, &queued, std::time::Duration::from_mins(2)).await;
|
|
}
|
|
for &container in infra {
|
|
let name = container.name();
|
|
match crate::priv_client::control_infra_container(container, InfraAction::Stop).await {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "stop: infra stop failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Best-effort server-side wait for a set of DAGs to settle terminal,
|
|
/// bounded by `timeout` — used to preserve ordering invariants inside
|
|
/// one request (agent stops before infra stops) without trusting the
|
|
/// client to wait.
|
|
async fn await_dags(coord: &Arc<Coordinator>, ids: &[u64], timeout: std::time::Duration) {
|
|
let deadline = std::time::Instant::now() + timeout;
|
|
loop {
|
|
// Every node under the named roots, terminal ones included — unlike the
|
|
// old typed snapshot, a settled group does not drop out of this view.
|
|
// So "pending" is simply "some node hasn't finished", with no second
|
|
// rule for the disappeared case.
|
|
let pending = coord
|
|
.job_queue
|
|
.node_subtrees(ids)
|
|
.iter()
|
|
.any(|n| !n.state.is_terminal());
|
|
if !pending {
|
|
return;
|
|
}
|
|
if std::time::Instant::now() >= deadline {
|
|
tracing::warn!(?ids, "await_dags: timed out; proceeding");
|
|
return;
|
|
}
|
|
tokio::time::sleep(std::time::Duration::from_millis(250)).await;
|
|
}
|
|
}
|
|
|
|
/// Start the given `infra` containers then `agents` (`hivectl start`) — the
|
|
/// inverse of [`handle_stop`]. Infra comes up before agents so the agents
|
|
/// find forge/matrix/gateway ready. Per-target failures aggregated. Callers
|
|
/// resolve the [`LifecycleScope`] to these explicit name lists up front.
|
|
async fn handle_start(
|
|
coord: &Arc<Coordinator>,
|
|
agents: &[String],
|
|
infra: &[InfraContainer],
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?agents, ?infra, "start");
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
|
|
for &container in infra {
|
|
let name = container.name();
|
|
match crate::priv_client::control_infra_container(container, InfraAction::Start).await {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "start: infra start failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
// One DAG for all targeted agents — a per-agent start subgraph each
|
|
// (`SetWanted(Up) → Reconcile`, or a rebuild-then-start for a stale
|
|
// rev), independent roots that run concurrently on their own leases. A
|
|
// hive-wide `hivectl start` is now a single DAG, not N. Through the
|
|
// queue: persists `wanted = Up`, per-agent stale-rev upgrade to a full
|
|
// rebuild, serializes on each agent's lease. The id rides back for
|
|
// hivectl's wait loop.
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::start_many(coord, agents).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue start: {e}")),
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Restart containers hive-wide (`hivectl restart`) — the DAG-based
|
|
/// sibling of [`handle_stop`]/[`handle_start`]. All targeted agents ride
|
|
/// **one** DAG (a per-agent restart subgraph each: `SetWanted → [Signal →
|
|
/// Drain →] StopForUpdate → Reconcile`, independent roots that run
|
|
/// concurrently on their own leases), not N separate DAGs — a hive-wide
|
|
/// restart is one job. `graceful` prepends signal→drain per agent. No
|
|
/// client-side stop-then-start composition, so a dropped `hivectl`
|
|
/// connection never strands an agent. Infra containers have no lease/DAG
|
|
/// and restart synchronously (stop then start).
|
|
async fn handle_restart_scoped(
|
|
coord: &Arc<Coordinator>,
|
|
scope: &LifecycleScope,
|
|
graceful: bool,
|
|
) -> Result<HostResponse> {
|
|
tracing::info!(?scope, graceful, "restart");
|
|
let agents = scoped_agents(scope).await?;
|
|
let infra = scoped_infra(scope);
|
|
let mut ok_items: Vec<String> = Vec::new();
|
|
let mut errors: Vec<String> = Vec::new();
|
|
let mut queued: Vec<u64> = Vec::new();
|
|
|
|
// One insert for all targeted agents — a per-agent restart subgraph each
|
|
// (`[Signal → Drain →] StopForUpdate → Reconcile`; no `SetWanted`, since a
|
|
// restart converges to the agent's *existing* intent rather than rewriting
|
|
// it), independent roots running concurrently on their own leases. No
|
|
// client-side stop-then-start composition — the whole restart survives a
|
|
// dropped connection because the graph owns it.
|
|
if !agents.is_empty() {
|
|
match crate::job_queue::power::restart_many(coord, &agents, graceful).await {
|
|
Ok(ids) => {
|
|
queued.extend(ids.into_iter().map(hive_jobq::NodeId::get));
|
|
ok_items.extend(agents.iter().cloned());
|
|
}
|
|
Err(e) => errors.push(format!("queue restart: {e}")),
|
|
}
|
|
}
|
|
|
|
for &container in &infra {
|
|
let name = container.name();
|
|
let res = async {
|
|
crate::priv_client::control_infra_container(container, InfraAction::Stop).await?;
|
|
crate::priv_client::control_infra_container(container, InfraAction::Start).await
|
|
}
|
|
.await;
|
|
match res {
|
|
Ok(()) => ok_items.push(name.to_owned()),
|
|
Err(e) => {
|
|
tracing::warn!(%name, error = ?e, "restart: infra restart failed");
|
|
errors.push(format!("{name}: {e:#}"));
|
|
}
|
|
}
|
|
}
|
|
|
|
let mut resp = finish_lifecycle(ok_items, &errors);
|
|
resp.queued_dags = Some(queued);
|
|
Ok(resp)
|
|
}
|
|
|
|
/// Resolve which sub-agent logical names a scope targets: every live
|
|
/// container (from `lifecycle::list`) when `agents` is set or the scope is
|
|
/// "everything", plus any explicit `agent_names`. Returns de-duplicated
|
|
/// logical names with the `h-` container prefix stripped.
|
|
/// A scope that targets *every* agent rather than an explicit
|
|
/// `--agent <name>` list: either the `agents` flag or a bare
|
|
/// "everything" scope. Broad scopes are the ones whose stop/start pair
|
|
/// drives the previously-running restore set (see `handle` Stop/Start
|
|
/// arms); a targeted `--agent` stop/start must not redefine it.
|
|
fn is_broad_scope(scope: &LifecycleScope) -> bool {
|
|
scope.agents || scope.is_everything()
|
|
}
|
|
|
|
/// Assemble this hive's domain + browser-facing web URLs from c0re's
|
|
/// service env (injected by the hyperhive NixOS module). Each field is `None` when its
|
|
/// surface isn't browser-reachable (domain unset, forge not behind the
|
|
/// gateway, matrix GUI off), so the CLI can hint precisely instead of
|
|
/// opening a dead link. Scheme matches the existing `HIVE_FORGE_PUBLIC_URL`
|
|
/// convention (gateway terminates TLS, so https).
|
|
fn hive_urls() -> hive_host_sock::HiveUrls {
|
|
// Treat an empty env value as unset everywhere — an empty domain would
|
|
// otherwise render `swarm.hives."" = …` (invalid nix) and `https:///`.
|
|
let env = |k: &str| std::env::var(k).ok().filter(|v| !v.is_empty());
|
|
let domain = env("HYPERHIVE_HIVE_DOMAIN");
|
|
hive_host_sock::HiveUrls {
|
|
home: domain.as_ref().map(|d| format!("https://{d}/")),
|
|
forge: env("HIVE_FORGE_PUBLIC_URL"),
|
|
matrix: env("HIVE_MATRIX_PUBLIC_URL"),
|
|
domain,
|
|
}
|
|
}
|
|
|
|
async fn scoped_agents(scope: &LifecycleScope) -> Result<Vec<String>> {
|
|
use std::collections::BTreeSet;
|
|
let mut set: BTreeSet<String> = BTreeSet::new();
|
|
if is_broad_scope(scope) {
|
|
for c in lifecycle::list().await? {
|
|
let logical = c
|
|
.strip_prefix(lifecycle::AGENT_PREFIX)
|
|
.unwrap_or(&c)
|
|
.to_owned();
|
|
set.insert(logical);
|
|
}
|
|
}
|
|
for n in &scope.agent_names {
|
|
set.insert(n.clone());
|
|
}
|
|
Ok(set.into_iter().collect())
|
|
}
|
|
|
|
/// Resolve which infra containers a scope targets. An "everything" scope
|
|
/// (no flags set) selects all controllable infra; otherwise each set flag
|
|
/// maps to its [`InfraContainer`]. Fixed order for deterministic output.
|
|
fn scoped_infra(scope: &LifecycleScope) -> Vec<InfraContainer> {
|
|
let everything = scope.is_everything();
|
|
let mut out = Vec::new();
|
|
if everything || scope.ci {
|
|
out.push(InfraContainer::Ci);
|
|
}
|
|
if everything || scope.forge {
|
|
out.push(InfraContainer::Forge);
|
|
}
|
|
if everything || scope.gateway {
|
|
out.push(InfraContainer::Gateway);
|
|
}
|
|
if everything || scope.matrix {
|
|
out.push(InfraContainer::Matrix);
|
|
}
|
|
out
|
|
}
|
|
|
|
/// Build the aggregated lifecycle response: `ok` with the touched names when
|
|
/// every target succeeded, otherwise `ok: false` with the joined errors and
|
|
/// the partial success list.
|
|
fn finish_lifecycle(ok_items: Vec<String>, errors: &[String]) -> HostResponse {
|
|
if errors.is_empty() {
|
|
HostResponse::list(ok_items)
|
|
} else {
|
|
HostResponse {
|
|
ok: false,
|
|
error: Some(errors.join("; ")),
|
|
agents: Some(ok_items),
|
|
..HostResponse::default()
|
|
}
|
|
}
|
|
}
|