From c6d9f59c4d0f6e16beb5cfef6d43b0f3635a6c55 Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 9 Jun 2026 11:34:49 +0200 Subject: [PATCH] refactor(#1474): extract dispatch arm logic into helpers (server + agent_server) --- hive-c0re/src/agent_server.rs | 656 +++++++++++++++++++--------------- hive-c0re/src/server.rs | 216 +++++------ 2 files changed, 476 insertions(+), 396 deletions(-) diff --git a/hive-c0re/src/agent_server.rs b/hive-c0re/src/agent_server.rs index 0a6345c1..f7662c55 100644 --- a/hive-c0re/src/agent_server.rs +++ b/hive-c0re/src/agent_server.rs @@ -115,8 +115,11 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> 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, + agent: &str, + wait_seconds: Option, + max: Option, +) -> 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, + 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, 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, + 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) -> 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 = - 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 = 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) -> } } -/// 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 { + 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, + 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, + 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, + 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, 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, 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, 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, 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 = 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 = 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, + agent: &str, + name: &str, + description: Option, +) -> 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, + 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, } +/// 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, diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 0c3e0d85..69e643d6 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -73,42 +73,10 @@ async fn handle(stream: UnixStream, coord: Arc) -> 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) -> HostResponse { let result: anyhow::Result = 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) -> 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 = Vec::new(); - let mut errors: Vec = 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) -> 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, name: &str) -> Result { + 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, name: &str) -> Result { + 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 { + tracing::info!("restart-all"); + let agents = lifecycle::list().await?; + let mut ok_agents: Vec = Vec::new(); + let mut errors: Vec = 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, name: &str) -> Result { + 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()) +}