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) -> 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) -> 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 resp = match serde_json::from_str::(line.trim()) { Ok(req) => dispatch(&req, coord.clone()).await, Err(e) => HostResponse::error(format!("parse error: {e}")), }; let mut payload = serde_json::to_string(&resp)?; payload.push('\n'); write.write_all(payload.as_bytes()).await?; write.flush().await?; } } #[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) -> HostResponse { let result: anyhow::Result = 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, // The hive domain + per-surface public URLs are injected into // c0re's service env by hive-c0re.nix; 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, name: &str) -> Result { 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)?; let _ = coord .push_todo( hive_sh4re::manager::MANAGER_AGENT, "core", Some(format!("spawned:{name}")), format!("agent '{name}' spawned"), None, false, ) .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 manager and propagate. let _ = coord .push_todo( hive_sh4re::manager::MANAGER_AGENT, "core", Some(format!("spawned:{name}")), format!("agent '{name}' spawn FAILED: {e:#}"), None, false, ) .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, 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) -> HostResponse { let rows = crate::container_view::build_all(&coord.hive_env()) .await .into_iter() .map(|v| hive_sh4re::container::AgentStatusRow { name: v.name, running: v.running, failed: v.failed, needs_update: v.needs_update, needs_login: v.needs_login, deployed_sha: v.deployed_sha, // Reminders are agent-local now; c0re has no cross-agent // visibility into pending counts anymore. Stubbed // to 0 rather than deleting the wire field outright — leaves // `hivectl status`/the dashboard column intact syntactically, // just always empty, until iris's frontend follow-up decides // whether to drop the column entirely. pending_reminders: 0, parent: v.parent, paused: v.paused, }) .collect(); HostResponse::agent_statuses(rows) } // --------------------------------------------------------------------------- // 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 + admin 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::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 { 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, name: &hive_types::Ident, cpu_quota: Option<&str>, memory_max: Option<&str>, ) -> Result { if let Some(value) = cpu_quota { crate::resource_limits::validate_cpu_quota(value).map_err(anyhow::Error::msg)?; } if let Some(value) = memory_max { crate::resource_limits::validate_memory_max(value).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: memory_max.map(ToOwned::to_owned), }; // 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.swarm.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 { require_matrix_present()?; let register_token = crate::matrix::ensure_register_token().context("read matrix register 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(), ®ister_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(), ®ister_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 { 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 { 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 { 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) -> Result { 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 { let agents: Vec = 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 { 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 { 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 { 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 { 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 { 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 { require_matrix_present()?; let register_token = crate::matrix::ensure_register_token().context("read matrix register token")?; let client = matrix_http_client()?; crate::matrix::ensure_admin_user(&client, ®ister_token) .await .context("matrix sync-admin")?; let path = crate::matrix::admin_token_path(); Ok(HostResponse::messages(vec![ format!( "matrix: hive admin user '@{}' provisioned", crate::matrix::HIVE_ADMIN_LOCALPART ), format!("token persisted at: {}", path.display()), ])) } async fn handle_matrix_promote_user(name: &str) -> Result { require_matrix_present()?; let admin_token = crate::matrix::read_admin_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, &admin_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 { require_matrix_present()?; let admin_token = crate::matrix::read_admin_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, &admin_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 { require_matrix_present()?; let admin_token = crate::matrix::read_admin_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, &admin_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, 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, agents: &[String], infra: &[InfraContainer], graceful: bool, ) -> Result { tracing::info!(?agents, ?infra, graceful, "stop"); let mut ok_items: Vec = Vec::new(); let mut errors: Vec = Vec::new(); let mut queued: Vec = 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, 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, agents: &[String], infra: &[InfraContainer], ) -> Result { tracing::info!(?agents, ?infra, "start"); let mut ok_items: Vec = Vec::new(); let mut errors: Vec = 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 = 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, scope: &LifecycleScope, graceful: bool, ) -> Result { tracing::info!(?scope, graceful, "restart"); let agents = scoped_agents(scope).await?; let infra = scoped_infra(scope); let mut ok_items: Vec = Vec::new(); let mut errors: Vec = Vec::new(); let mut queued: Vec = 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 ` 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 hive-c0re.nix). 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> { use std::collections::BTreeSet; let mut set: BTreeSet = 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 { 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, 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() } } }