hyperhive/hive-c0re/src/socket_server/lifecycle_handlers.rs

//! Container-lifecycle request handlers (`Start` / `Restart` / `Kill` /
//! `Update` / `ListDescendants`), including the capability-gated
//! infra-container restart path. All are topology-guarded via
//! `super::require_descendant`.

use std::sync::Arc;

use hive_core_agent_sock::Response;

use super::require_descendant;
use crate::coordinator::Coordinator;

/// `Start` — start a container, kicking its next turn. The caller must be an
/// ancestor of `name` in the topology (the root covers every agent).
pub(super) async fn handle_start(coord: &Arc<Coordinator>, agent: &str, name: &str) -> Response {
    if let Some(err) = require_descendant(agent, name, "start") {
        return err;
    }
    tracing::info!(%agent, %name, "start container");
    // Persist `wanted = Up` and submit the Start DAG; the submit layer
    // upgrades a stale-rev start to a full rebuild so the container
    // runs current nix derivations before it starts.
    if let Err(e) = crate::job_queue::power::start_many(coord, &[name.to_owned()]).await {
        tracing::error!(%agent, %name, error = ?e, "start: insert failed");
    }
    Response::Ok
}

/// `Restart` — enqueue a restart for a container. The caller must be an
/// ancestor of `name` in the topology. The infra-container branch is
/// orthogonal: it is gated on the `infra_admin` capability and audited, so it
/// stays ahead of the topology guard.
pub(super) async fn handle_restart(coord: &Arc<Coordinator>, agent: &str, name: &str) -> Response {
    // Infra restart: an agent holding the `infra_admin` capability can
    // restart a hive infrastructure service (hive-ci / hive-forge /
    // hive-matrix) by passing its name to the same restart tool. The
    // `InfraContainer` enum parse both recognises these (never agent
    // children, so disjoint from the child path below) and yields the typed
    // value the restart path needs. It recognises `hive-gateway` too, which
    // is then refused — a name the agent surface knows but may not act on.
    if let Ok(container) = name.parse::<hive_priv_sock::InfraContainer>() {
        return handle_restart_infra(coord, agent, container).await;
    }
    if let Some(err) = require_descendant(agent, name, "restart") {
        return err;
    }
    tracing::info!(%agent, %name, "submit restart");
    if let Err(e) = crate::job_queue::power::restart_many(coord, &[name.to_owned()], false).await {
        tracing::error!(%agent, %name, error = ?e, "restart: insert failed");
    }
    Response::Ok
}

/// Restart a hive infrastructure container on behalf of an agent that
/// holds the `infra_admin` capability. The `container` is already a valid
/// [`hive_priv_sock::InfraContainer`] (the caller parsed it); this gates on the capability
/// and routes the systemctl restart through hive-priv. Direct, not
/// approval-gated.
async fn handle_restart_infra(
    coord: &Arc<Coordinator>,
    agent: &str,
    container: hive_priv_sock::InfraContainer,
) -> Response {
    let name = container.name();
    // Record the attempt in the operator-visible privileged-action audit
    // trail, then emit a live `AuditEntryAdded` so the dashboard audit view
    // appends it off `/dashboard/stream`. Best-effort: `record` returns the
    // canonical row (or `None` on a sqlite blip), and we stream exactly that
    // row so the stored + streamed views can't drift. `action` is stable so
    // the dashboard can group/filter.
    let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| {
        if let Some(entry) = coord
            .audit_log
            .record(agent, "restart_infra", name, outcome, detail)
        {
            coord.emit_audit_entry(entry);
        }
    };
    // Some targets are off-limits to agents regardless of capability — the
    // gateway, because nginx fronts every hive service from the host and an
    // agent bouncing it takes out the forge, the dashboard and matrix at
    // once, including the route its own fix would have to travel. Checked
    // before the capability so the refusal doesn't read as "ask for
    // infra_admin"; no capability grants this.
    if !container.agent_restartable() {
        tracing::warn!(%agent, %name, "agent: infra restart denied (not agent-restartable)");
        audit(
            crate::audit_log::AuditOutcome::Err,
            Some("denied: target is not agent-restartable"),
        );
        return Response::Err {
            message: format!("`{name}` cannot be restarted by an agent; ask the operator"),
        };
    }
    if !crate::capabilities::has_cap(agent, hive_sh4re::permissions::Capability::InfraAdmin) {
        tracing::warn!(%agent, %name, "agent: infra restart denied (no infra_admin capability)");
        audit(
            crate::audit_log::AuditOutcome::Err,
            Some("denied: missing infra_admin capability"),
        );
        return Response::Err {
            message: format!(
                "restarting infra container `{name}` requires the `infra_admin` capability"
            ),
        };
    }
    tracing::info!(%agent, %name, "agent: restart infra container");
    match crate::priv_client::restart_infra_container(container).await {
        Ok(()) => {
            audit(crate::audit_log::AuditOutcome::Ok, None);
            Response::Ok
        }
        Err(e) => {
            let msg = format!("{e:#}");
            audit(crate::audit_log::AuditOutcome::Err, Some(&msg));
            Response::Err { message: msg }
        }
    }
}

/// `Kill` — kill a container, unregister it, notify the manager. The caller
/// must be an ancestor of `name` in the topology.
pub(super) async fn handle_kill(coord: &Arc<Coordinator>, agent: &str, name: &str) -> Response {
    if let Some(err) = require_descendant(agent, name, "kill") {
        return err;
    }
    tracing::info!(%agent, %name, "kill container");
    // Persist the intent even if the kill fails — otherwise the next
    // reconcile would restart the container.
    if let Err(e) = coord.power.set(name, crate::power::Wanted::Offline) {
        tracing::warn!(%name, error = ?e, "agent_power: set wanted=offline failed");
    }
    let result: anyhow::Result<()> = async {
        crate::lifecycle::kill(name).await?;
        coord.unregister_agent(name);
        Ok(())
    }
    .await;
    match result {
        Ok(()) => {
            let _ = coord
                .push_todo(
                    hive_sh4re::manager::MANAGER_AGENT,
                    "core",
                    Some(format!("killed:{name}")),
                    format!("agent '{name}' killed"),
                    None,
                    false,
                )
                .await;
            Response::Ok
        }
        Err(e) => Response::Err {
            message: format!("{e:#}"),
        },
    }
}

/// `Update` — enqueue a rebuild for a container. The caller must be an
/// ancestor of `name` in the topology.
pub(super) fn handle_update(coord: &Arc<Coordinator>, agent: &str, name: &str) -> Response {
    if let Some(err) = require_descendant(agent, name, "rebuild") {
        return err;
    }
    tracing::info!(%agent, %name, "submit rebuild");
    if let Err(e) = coord.job_queue.insert_job(|b| {
        crate::job_queue::templates::rebuild(b, name, true);
        Vec::new()
    }) {
        tracing::error!(%agent, %name, error = ?e, "update: insert failed");
    }
    coord.emit_rebuild_queue_snapshot();
    Response::Ok
}

/// `ListDescendants` — every topological descendant of `agent` with
/// its running/stopped state, parents before children.
pub(super) async fn handle_list_descendants(coord: &Arc<Coordinator>, agent: &str) -> Response {
    tracing::debug!(%agent, "agent: list descendants");
    // Walk the full topology and collect every descendant.
    let topo = crate::topology::read();
    let mut names: Vec<String> = topo
        .keys()
        .filter(|name| crate::topology::is_descendant_of(name, agent))
        .cloned()
        .collect();
    // Parents before children, then alpha within each tier.
    crate::auto_update::topology_sort(&mut names, &topo);
    // Read from the coordinator's cached container snapshot instead of
    // live-querying each container's systemd unit state — the same
    // `containers_snapshot()` the dashboard's `/api/state` cold-load path
    // already uses, kept fresh by `rescan_containers_and_emit()` on every
    // mutation plus the crash-watcher's periodic poll. Avoids N
    // `systemctl is-active` subprocess spawns per `list_containers` call;
    // per mara, daemons should do the expensive work themselves and serve
    // clients a cheap cached read.
    let snapshot = coord.containers_snapshot().await;
    let running_by_name: std::collections::HashMap<&str, bool> = snapshot
        .iter()
        .map(|v| (v.name.as_str(), v.running))
        .collect();
    let containers = names
        .into_iter()
        .map(|name| {
            // A descendant absent from the snapshot (not yet scanned since
            // its own registration, e.g. mid-spawn) reads as not running
            // rather than erroring — matches the old membership-check's
            // default-false behavior for an unknown name.
            let running = running_by_name.get(name.as_str()).copied().unwrap_or(false);
            hive_sh4re::container::ContainerInfo { name, running }
        })
        .collect();
    Response::Containers { containers }
}