vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

refactor(#1474): extract dispatch arm logic into helpers (server + agent_server)

Browse source
This commit is contained in:
damocles 2026-06-09 11:34:49 +02:00 • committed by mara
commit c6d9f59c4d
2 changed files with 476 additions and 396 deletions
Download patch file Download diff file Expand all files Collapse all files

656
hive-c0re/src/agent_server.rs
View file

@ -115,8 +115,11 @@ pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
#[allow(
clippy::too_many_lines,
reason = "flat dispatch table: one match arm per shared request variant; \
splitting it would scatter the routing logic without shrinking it"
reason = "flat routing table over the shared request variants — the \
logic-bearing arms (recv / wake / set_status / get_agent_meta) \
are extracted to handlers; what's left is one short broker \
dispatch per variant, kept inline so the routing map reads \
top-to-bottom"
)]
/// Handle the subset of `Request` variants that are identical on both
/// the agent socket and the manager socket. Returns `Some(response)` for
@ -139,27 +142,7 @@ pub(crate) async fn dispatch_shared(
in_reply_to,
} => handle_send(coord, agent, to, body, *in_reply_to),
hive_sh4re::Request::Recv { wait_seconds, max } => {
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match broker
.recv_blocking_batch(agent, recv_timeout(*wait_seconds), cap)
.await
{
Ok(deliveries) => hive_sh4re::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::DeliveredMessage {
from: d.message.from,
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
},
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
handle_recv(coord, agent, *wait_seconds, *max).await
}
hive_sh4re::Request::Status => match broker.count_pending(agent) {
Ok(unread) => hive_sh4re::Response::Status { unread },
@ -182,27 +165,7 @@ pub(crate) async fn dispatch_shared(
from,
body,
transient,
} => {
if *transient {
// Transient wakes bypass sqlite — they fire the broadcast
// channel only. No redelivery on restart; no message history
// entry. Used by bash task completions.
broker.ping(agent, from, body);
hive_sh4re::Response::Ok
} else {
match broker.send(&Message {
from: from.clone(),
to: agent.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
}
} => handle_wake(coord, agent, from, body, *transient),
hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) {
Ok(rows) => hive_sh4re::Response::Recent { rows },
Err(e) => hive_sh4re::Response::Err {
@ -239,32 +202,9 @@ pub(crate) async fn dispatch_shared(
timing,
file_path,
} => handle_remind(coord, agent, message, timing, file_path.as_deref()),
hive_sh4re::Request::SetStatus { text } => {
if let Err(message) = crate::limits::check_status_text(text) {
return Some(hive_sh4re::Response::Err { message });
}
// The harness writes the status file to its own `state/` dir
// before sending this request (it runs as the agent user, so
// it has write access). We just trigger a dashboard rescan so
// the new value is reflected immediately.
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok
}
hive_sh4re::Request::SetStatus { text } => handle_set_status(coord, text),
hive_sh4re::Request::GetAgentMeta { name } => {
let target = name.as_deref().unwrap_or(agent);
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
hive_sh4re::Response::AgentMeta {
name: target.to_owned(),
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
handle_get_agent_meta(coord, agent, name.as_deref()).await
}
hive_sh4re::Request::CancelLooseEnd { kind, id } => {
crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
@ -317,247 +257,140 @@ pub(crate) async fn dispatch_shared(
})
}
#[allow(
clippy::too_many_lines,
reason = "flat dispatch table: one match arm per agent-socket request \
variant; splitting it would scatter the routing logic"
)]
/// `Recv` — long-poll the broker for up to `max` messages (capped at
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
async fn handle_recv(
coord: &Arc<Coordinator>,
agent: &str,
wait_seconds: Option<u64>,
max: Option<u32>,
) -> hive_sh4re::Response {
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
match coord
.broker
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
.await
{
Ok(deliveries) => hive_sh4re::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::DeliveredMessage {
from: d.message.from,
body: d.message.body,
id: d.id,
redelivered: d.redelivered,
in_reply_to: d.message.in_reply_to,
})
.collect(),
},
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes
/// fire the broadcast channel only (no sqlite row, no redelivery on
/// restart — used by bash-task completions); durable wakes persist
/// through the broker like any other message.
fn handle_wake(
coord: &Arc<Coordinator>,
agent: &str,
from: &str,
body: &str,
transient: bool,
) -> hive_sh4re::Response {
let broker = &coord.broker;
if transient {
broker.ping(agent, from, body);
hive_sh4re::Response::Ok
} else {
match broker.send(&Message {
from: from.to_owned(),
to: agent.to_owned(),
body: body.to_owned(),
in_reply_to: None,
}) {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
}
/// `SetStatus` — validate the status text, then trigger a dashboard
/// rescan. The harness has already written the status file to its own
/// `state/` dir (it runs as the agent user), so this only refreshes the
/// dashboard's view.
fn handle_set_status(coord: &Arc<Coordinator>, text: &str) -> hive_sh4re::Response {
if let Err(message) = crate::limits::check_status_text(text) {
return hive_sh4re::Response::Err { message };
}
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok
}
/// `GetAgentMeta` — identity + live status for `name` (defaults to the
/// caller). Reads the live container-view status and the hive/swarm
/// display names.
async fn handle_get_agent_meta(
coord: &Arc<Coordinator>,
agent: &str,
name: Option<&str>,
) -> hive_sh4re::Response {
let target = name.unwrap_or(agent);
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
hive_sh4re::Response::AgentMeta {
name: target.to_owned(),
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
if let Some(resp) = dispatch_shared(req, agent, coord).await {
return resp;
}
match req {
AgentRequest::GetLooseEnds { agent: target } => {
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match crate::loose_ends::for_agent(coord, name) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
handle_get_loose_ends(coord, agent, target.as_deref())
}
AgentRequest::CountPendingReminders { agent: target } => {
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match coord.broker.count_pending_reminders_for(name) {
Ok(count) => AgentResponse::PendingRemindersCount { count },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
handle_count_pending_reminders(coord, agent, target.as_deref())
}
AgentRequest::ReminderRollup {
since_secs,
agent: target,
} => {
let name = resolve_agent_state_target(agent, target.as_deref());
match name {
Ok(name) => match coord.broker.reminder_rollup_for(name, *since_secs) {
Ok(stats) => AgentResponse::ReminderRollup(stats),
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
AgentRequest::Start { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot start `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: start child");
match crate::lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::Restart { name } => {
// Topology check: the caller must be the direct parent of the
// target. This is the only authorisation criterion — no
// capability flag needed; parenthood is sufficient privilege.
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot restart `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: enqueue restart for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` restart tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
AgentRequest::Kill { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot kill `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: kill child");
let result: anyhow::Result<()> = async {
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.clone(),
});
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::Update { name } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot rebuild `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` update tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
AgentRequest::ListDescendants => {
tracing::debug!(%agent, "agent: list descendants");
// All containers known to nixos-container (running only).
let running_set: std::collections::HashSet<String> =
match crate::lifecycle::list().await {
Ok(names) => names
.into_iter()
.filter_map(|c| {
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
.map(str::to_owned)
})
.collect(),
Err(e) => {
return AgentResponse::Err {
message: format!("list containers failed: {e:#}"),
};
}
};
// 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);
let containers = names
.into_iter()
.map(|name| {
let running = running_set.contains(&name);
hive_sh4re::ContainerInfo { name, running }
})
.collect();
AgentResponse::Containers { containers }
}
} => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
AgentRequest::Start { name } => handle_start_child(coord, agent, name).await,
AgentRequest::Restart { name } => handle_restart_child(coord, agent, name),
AgentRequest::Kill { name } => handle_kill_child(coord, agent, name).await,
AgentRequest::Update { name } => handle_update_child(coord, agent, name),
AgentRequest::ListDescendants => handle_list_descendants(agent).await,
AgentRequest::RequestInitConfig { name, description } => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == name)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot request_init_config for `{name}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %name, "agent: request_init_config for child");
match crate::manager_server::submit_init_config(coord, name, description.clone()) {
Ok(_id) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
handle_request_init_config(coord, agent, name, description.clone())
}
AgentRequest::RequestApplyCommit {
agent: target_agent,
commit_ref,
description,
} => {
if !crate::topology::children_of(agent)
.iter()
.any(|c| c == target_agent)
{
return AgentResponse::Err {
message: format!(
"agent `{agent}` cannot request_apply_commit for `{target_agent}`: \
not a direct child in the topology tree"
),
};
}
tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
match crate::manager_server::submit_apply_commit(
handle_request_apply_commit(
coord,
agent,
target_agent,
commit_ref,
description.as_deref(),
)
.await
{
Ok((id, sha)) => {
tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
// Manager-only variants are not valid on the agent socket.
_ => AgentResponse::Err {
@ -566,13 +399,245 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
}
}
/// Handle `GetHostJournal` from both the agent and manager sockets.
/// Capability-gated: the calling agent must hold `read_host_journal` in
/// `meta/capabilities.json`. Runs `journalctl` host-side and returns
/// the output as a `HostJournal` response.
///
/// The manager is not exempt - grant `read_host_journal` in
/// `meta/capabilities.json` to enable it for any agent including the manager.
/// Topology guard for the agent-socket lifecycle/config tools: the
/// caller must be the direct parent of `target`. Returns `Some(Err)`
/// to short-circuit the dispatch arm when it isn't, `None` when the
/// call is authorised. `action` is the verb phrase for the message
/// (e.g. `"start"`, `"request_apply_commit for"`).
fn require_child(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
if crate::topology::children_of(agent)
.iter()
.any(|c| c == target)
{
None
} else {
Some(AgentResponse::Err {
message: format!(
"agent `{agent}` cannot {action} `{target}`: \
not a direct child in the topology tree"
),
})
}
}
/// `GetLooseEnds` — resolve the (optionally cross-agent) target then
/// read its loose ends.
fn handle_get_loose_ends(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match crate::loose_ends::for_agent(coord, name) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `CountPendingReminders` — resolve the target then count its pending
/// reminders.
fn handle_count_pending_reminders(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match coord.broker.count_pending_reminders_for(name) {
Ok(count) => AgentResponse::PendingRemindersCount { count },
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `ReminderRollup` — resolve the target then roll up its reminders
/// fired in the last `since_secs`.
fn handle_reminder_rollup(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
since_secs: u64,
) -> AgentResponse {
match resolve_agent_state_target(agent, target) {
Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
Ok(stats) => AgentResponse::ReminderRollup(stats),
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
}
}
/// `Start` — start a direct-child container, kicking its next turn.
async fn handle_start_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "start") {
return err;
}
tracing::info!(%agent, %name, "agent: start child");
match crate::lifecycle::start(name).await {
Ok(()) => {
coord.kick_agent(name, "container started");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Restart` — enqueue a restart for a direct-child container.
/// Topology parenthood is the only authorisation criterion — no
/// capability flag needed.
fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "restart") {
return err;
}
tracing::info!(%agent, %name, "agent: enqueue restart for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Restart,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` restart tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
/// `Kill` — kill a direct-child container, unregister it, notify the
/// manager.
async fn handle_kill_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "kill") {
return err;
}
tracing::info!(%agent, %name, "agent: kill child");
let result: anyhow::Result<()> = async {
crate::lifecycle::kill(name).await?;
coord.unregister_agent(name);
Ok(())
}
.await;
match result {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.to_owned(),
});
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `Update` — enqueue a rebuild for a direct-child container.
fn handle_update_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
if let Some(err) = require_child(agent, name, "rebuild") {
return err;
}
tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
coord.rebuild_queue.enqueue(
crate::rebuild_queue::QueueKind::Rebuild,
name.to_owned(),
crate::rebuild_queue::QueueSource::Manual,
format!("agent `{agent}` update tool"),
None,
);
coord.emit_rebuild_queue_snapshot();
AgentResponse::Ok
}
/// `ListDescendants` — every topological descendant of `agent` with
/// its running/stopped state, parents before children.
async fn handle_list_descendants(agent: &str) -> AgentResponse {
tracing::debug!(%agent, "agent: list descendants");
// All containers known to nixos-container (running only).
let running_set: std::collections::HashSet<String> = match crate::lifecycle::list().await {
Ok(names) => names
.into_iter()
.filter_map(|c| {
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
.map(str::to_owned)
})
.collect(),
Err(e) => {
return AgentResponse::Err {
message: format!("list containers failed: {e:#}"),
};
}
};
// 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);
let containers = names
.into_iter()
.map(|name| {
let running = running_set.contains(&name);
hive_sh4re::ContainerInfo { name, running }
})
.collect();
AgentResponse::Containers { containers }
}
/// `RequestInitConfig` — queue an `InitConfig` approval for a
/// direct-child agent.
fn handle_request_init_config(
coord: &Arc<Coordinator>,
agent: &str,
name: &str,
description: Option<String>,
) -> AgentResponse {
if let Some(err) = require_child(agent, name, "request_init_config for") {
return err;
}
tracing::info!(%agent, %name, "agent: request_init_config for child");
match crate::manager_server::submit_init_config(coord, name, description) {
Ok(_id) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// `RequestApplyCommit` — queue an apply-commit approval for a
/// direct-child agent.
async fn handle_request_apply_commit(
coord: &Arc<Coordinator>,
agent: &str,
target_agent: &str,
commit_ref: &str,
description: Option<&str>,
) -> AgentResponse {
if let Some(err) = require_child(agent, target_agent, "request_apply_commit for") {
return err;
}
tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
match crate::manager_server::submit_apply_commit(coord, target_agent, commit_ref, description)
.await
{
Ok((id, sha)) => {
tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
}
}
/// Field-named journal-query knobs for [`dispatch_host_journal`].
/// Borrows straight from the matched `GetHostJournal` request variant.
pub struct HostJournalArgs<'a> {
@ -585,6 +650,13 @@ pub struct HostJournalArgs<'a> {
pub until: &'a Option<String>,
}
/// Handle `GetHostJournal` from both the agent and manager sockets.
/// Capability-gated: the calling agent must hold `read_host_journal` in
/// `meta/capabilities.json`. Runs `journalctl` host-side and returns
/// the output as a `HostJournal` response.
///
/// The manager is not exempt - grant `read_host_journal` in
/// `meta/capabilities.json` to enable it for any agent including the manager.
pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> AgentResponse {
let HostJournalArgs {
unit,

216
hive-c0re/src/server.rs
View file

@ -73,42 +73,10 @@ async fn handle(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
}
}
#[allow(
clippy::too_many_lines,
reason = "flat dispatch table: one match arm per host-socket request \
variant; splitting it would scatter the routing logic"
)]
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 hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
match lifecycle::spawn(name, &hive, &paths).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::Spawn { name } => handle_spawn(&coord, name).await?,
HostRequest::RequestSpawn { name } => {
tracing::info!(%name, "request_spawn");
let id =
@ -118,86 +86,18 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
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::Kill { name } => handle_kill(&coord, name).await?,
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 ok_agents: Vec<String> = Vec::new();
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:#}"));
} else {
ok_agents.push(agent.clone());
}
}
if errors.is_empty() {
HostResponse::list(ok_agents)
} else {
HostResponse {
ok: false,
error: Some(errors.join("; ")),
agents: Some(ok_agents),
approvals: None,
}
}
}
HostRequest::RestartAll => handle_restart_all().await?,
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 hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).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::Rebuild { name } => handle_rebuild(&coord, name).await?,
HostRequest::List => HostResponse::list(lifecycle::list().await?),
HostRequest::Pending => HostResponse::pending(coord.approvals.pending()?),
HostRequest::Approve { id } => {
@ -229,3 +129,111 @@ async fn dispatch(req: &HostRequest, coord: Arc<Coordinator>) -> HostResponse {
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 = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
match lifecycle::spawn(name, &hive, &paths).await {
Ok(()) => {
coord.notify_manager(&hive_sh4re::HelperEvent::Spawned {
agent: name.to_owned(),
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.to_owned(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
});
return Err(e);
}
}
Ok(HostResponse::success())
}
/// Kill `name`'s container, unregister its socket, notify the manager.
async fn handle_kill(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
tracing::info!(%name, "kill");
lifecycle::kill(name).await?;
coord.unregister_agent(name);
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
agent: name.to_owned(),
});
Ok(HostResponse::success())
}
/// Restart every container, aggregating per-agent failures into one
/// response rather than aborting on the first error.
async fn handle_restart_all() -> Result<HostResponse> {
tracing::info!("restart-all");
let agents = lifecycle::list().await?;
let mut ok_agents: Vec<String> = Vec::new();
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:#}"));
} else {
ok_agents.push(agent.clone());
}
}
if errors.is_empty() {
Ok(HostResponse::list(ok_agents))
} else {
Ok(HostResponse {
ok: false,
error: Some(errors.join("; ")),
agents: Some(ok_agents),
approvals: None,
})
}
}
/// Rebuild `name`'s container, notifying the manager of the outcome
/// (success or failure) and kicking the agent's next turn on success.
async fn handle_rebuild(coord: &Arc<Coordinator>, name: &str) -> Result<HostResponse> {
tracing::info!(%name, "rebuild");
let agent_dir = coord.ensure_runtime(name)?;
let hive = coord.hive_env();
let paths = Coordinator::agent_paths(name, agent_dir);
let result = lifecycle::rebuild(name, &hive, &paths, &|_| (), &|_| ()).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.to_owned(),
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.to_owned(),
ok: false,
note: Some(format!("{e:#}")),
sha: None,
tag: None,
}),
}
result?;
Ok(HostResponse::success())
}