256 lines
10 KiB
Rust
256 lines
10 KiB
Rust
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,
|
|
¬es_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::Restart { name } => {
|
|
tracing::info!(%name, "restart");
|
|
lifecycle::restart(name).await?;
|
|
HostResponse::success()
|
|
}
|
|
HostRequest::RestartAll => {
|
|
tracing::info!("restart-all");
|
|
let agents = lifecycle::list().await?;
|
|
let mut errors: Vec<String> = Vec::new();
|
|
for agent in &agents {
|
|
if let Err(e) = lifecycle::restart(agent).await {
|
|
tracing::warn!(%agent, error = ?e, "restart-all: failed to restart agent");
|
|
errors.push(format!("{agent}: {e:#}"));
|
|
}
|
|
}
|
|
if errors.is_empty() {
|
|
HostResponse::list(agents)
|
|
} else {
|
|
HostResponse {
|
|
ok: false,
|
|
error: Some(errors.join("; ")),
|
|
agents: Some(agents),
|
|
approvals: None,
|
|
}
|
|
}
|
|
}
|
|
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,
|
|
¬es_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:#}")),
|
|
}
|
|
}
|