hyperhive/hive-c0re/src/server.rs

use std::path::Path;
use std::sync::Arc;

use anyhow::{Context, Result};
use hive_sh4re::{HostRequest, HostResponse};
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 resp = match serde_json::from_str::<HostRequest>(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)]
async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
    let result: anyhow::Result<HostResponse> = async {
        Ok(match req {
            HostRequest::Spawn { name } => {
                tracing::info!(%name, "spawn");
                let agent_dir = coord.ensure_runtime(name)?;
                let proposed_dir = Coordinator::agent_proposed_dir(name);
                let applied_dir = Coordinator::agent_applied_dir(name);
                let claude_dir = Coordinator::agent_claude_dir(name);
                let notes_dir = Coordinator::agent_notes_dir(name);
                match lifecycle::spawn(
                    name,
                    &coord.hyperhive_flake,
                    &coord.nixpkgs_flake,
                    &coord.nixpkgs_unstable_flake,
                    &agent_dir,
                    &proposed_dir,
                    &applied_dir,
                    &claude_dir,
                    &notes_dir,
                    coord.dashboard_port,
                    &coord.operator_pronouns,
                    &coord.context_window_tokens,
                )
                .await
                {
                    Ok(()) => {
                        coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
                            agent: name.clone(),
                            ok: true,
                            note: None,
                            sha: None,
                        });
                    }
                    Err(e) => {
                        // Roll back socket registration if container creation failed.
                        coord.unregister_agent(name);
                        coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
                            agent: name.clone(),
                            ok: false,
                            note: Some(format!("{e:#}")),
                            sha: None,
                        });
                        return Err(e);
                    }
                }
                HostResponse::success()
            }
            HostRequest::RequestSpawn { name } => {
                tracing::info!(%name, "request_spawn");
                let id =
                    coord
                        .approvals
                        .submit_kind(name, hive_sh4re::ApprovalKind::Spawn, "", None)?;
                tracing::info!(%id, %name, "spawn approval queued");
                HostResponse::success()
            }
            HostRequest::Kill { name } => {
                tracing::info!(%name, "kill");
                lifecycle::kill(name).await?;
                coord.unregister_agent(name);
                coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
                    agent: name.clone(),
                });
                HostResponse::success()
            }
            HostRequest::Destroy { name, purge } => {
                actions::destroy(&coord, name, *purge).await?;
                HostResponse::success()
            }
            HostRequest::Rebuild { name } => {
                tracing::info!(%name, "rebuild");
                let agent_dir = coord.ensure_runtime(name)?;
                let applied_dir = Coordinator::agent_applied_dir(name);
                let claude_dir = Coordinator::agent_claude_dir(name);
                let notes_dir = Coordinator::agent_notes_dir(name);
                let result = lifecycle::rebuild(
                    name,
                    &coord.hyperhive_flake,
                    &coord.nixpkgs_flake,
                    &coord.nixpkgs_unstable_flake,
                    &agent_dir,
                    &applied_dir,
                    &claude_dir,
                    &notes_dir,
                    coord.dashboard_port,
                    &coord.operator_pronouns,
                    &coord.context_window_tokens,
                    &|_| (),
                )
                .await;
                // Mirror auto_update::rebuild_agent — the manager wants
                // to know about every rebuild attempt regardless of
                // which surface triggered it, especially failures
                // (build error → manager can adjust the agent's
                // agent.nix). Without this the admin-socket CLI was
                // a notify-gap.
                match &result {
                    Ok(()) => {
                        coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                            agent: name.clone(),
                            ok: true,
                            note: None,
                            sha: None,
                            tag: None,
                        });
                        // Wake the agent's next turn with the
                        // "you were rebuilt" hint. Same pattern as
                        // auto_update::rebuild_agent and the dashboard
                        // rebuild path — this is the CLI's equivalent.
                        coord.kick_agent(name, "container rebuilt");
                    }
                    Err(e) => coord.notify_manager(&hive_sh4re::HelperEvent::Rebuilt {
                        agent: name.clone(),
                        ok: false,
                        note: Some(format!("{e:#}")),
                        sha: None,
                        tag: None,
                    }),
                }
                result?;
                HostResponse::success()
            }
            HostRequest::List => HostResponse::list(lifecycle::list().await?),
            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).await?;
                HostResponse::success()
            }
            HostRequest::SetParent { child, new_parent } => {
                tracing::info!(%child, ?new_parent, "set_parent");
                // `reparent_with_notify` wraps `topology::set_parent`
                // with the three notification messages + the
                // ContainerView rescan. Idempotent same-parent calls
                // skip both the messages and the disk write per the
                // topology fast-path.
                coord
                    .reparent_with_notify(child, new_parent.as_deref())
                    .await
                    .map_err(anyhow::Error::msg)?;
                HostResponse::success()
            }
        })
    }
    .await;
    match result {
        Ok(r) => r,
        Err(e) => HostResponse::error(format!("{e:#}")),
    }
}