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(#1019): unify agent + manager server dispatch via dispatch_shared

Browse source
This commit is contained in:
damocles 2026-06-01 21:03:17 +02:00
commit 68f488d81f
2 changed files with 112 additions and 332 deletions
Download patch file Download diff file Expand all files Collapse all files

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

@ -95,7 +95,7 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Re
/// cheap "is there anything pending?" check without blocking the
/// turn for 30 seconds. To actually park, the caller passes a
/// positive `wait_seconds`.
const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180);
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180);
/// Server-side hard cap on `Recv.max`. Bounds the size of a single
/// round-trip so a confused caller can't drain the entire inbox in
@ -104,29 +104,40 @@ const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(1
/// seen in practice (post-rebuild rescue, multi-agent reply storms)
/// and well under the per-message `MESSAGE_MAX_BYTES` * N envelope
/// budget.
const RECV_BATCH_MAX: u32 = 32;
pub(crate) const RECV_BATCH_MAX: u32 = 32;
fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
match wait_seconds {
Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX),
None => std::time::Duration::ZERO,
}
}
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
/// Handle the subset of `Request` variants that are identical on both
/// the agent socket and the manager socket. Returns `Some(response)` for
/// every variant it handles; returns `None` for variants with socket-specific
/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup`
/// where the manager can target other agents) or for manager-only variants.
///
/// Both `agent_server::dispatch` and `manager_server::dispatch` call this
/// first; each then handles its own remaining arms.
pub(crate) async fn dispatch_shared(
req: &hive_sh4re::Request,
agent: &str,
coord: &Arc<Coordinator>,
) -> Option<hive_sh4re::Response> {
let broker = &coord.broker;
match req {
AgentRequest::Send { to, body, in_reply_to } => {
Some(match req {
hive_sh4re::Request::Send { to, body, in_reply_to } => {
handle_send(coord, agent, to, body, *in_reply_to)
}
AgentRequest::Recv { wait_seconds, max } => {
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) => AgentResponse::Messages {
Ok(deliveries) => hive_sh4re::Response::Messages {
messages: deliveries
.into_iter()
.map(|d| hive_sh4re::DeliveredMessage {
@ -138,46 +149,46 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
})
.collect(),
},
Err(e) => AgentResponse::Err {
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
}
}
AgentRequest::Status => match broker.count_pending(agent) {
Ok(unread) => AgentResponse::Status { unread },
Err(e) => AgentResponse::Err {
hive_sh4re::Request::Status => match broker.count_pending(agent) {
Ok(unread) => hive_sh4re::Response::Status { unread },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
AgentRequest::OperatorMsg { body } => match broker.send(&Message {
hive_sh4re::Request::OperatorMsg { body } => match broker.send(&Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
to: agent.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
AgentRequest::Wake { from, body } => match broker.send(&Message {
hive_sh4re::Request::Wake { from, body } => match broker.send(&Message {
from: from.clone(),
to: agent.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
Ok(()) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
AgentRequest::Recent { limit } => match broker.recent_for(agent, *limit) {
Ok(rows) => AgentResponse::Recent { rows },
Err(e) => AgentResponse::Err {
hive_sh4re::Request::Recent { limit } => match broker.recent_for(agent, *limit) {
Ok(rows) => hive_sh4re::Response::Recent { rows },
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
AgentRequest::Ask {
hive_sh4re::Request::Ask {
question,
options,
multi,
@ -193,21 +204,108 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
to.as_deref(),
)
.map_or_else(
|message| AgentResponse::Err { message },
|id| AgentResponse::QuestionQueued { id },
|message| hive_sh4re::Response::Err { message },
|id| hive_sh4re::Response::QuestionQueued { id },
),
AgentRequest::Answer { id, answer } => crate::questions::handle_answer(
coord, agent, *id, answer,
)
.map_or_else(
|message| AgentResponse::Err { message },
|()| AgentResponse::Ok,
),
AgentRequest::Remind {
hive_sh4re::Request::Answer { id, answer } => {
crate::questions::handle_answer(coord, agent, *id, answer).map_or_else(
|message| hive_sh4re::Response::Err { message },
|()| hive_sh4re::Response::Ok,
)
}
hive_sh4re::Request::Remind {
message,
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 });
}
let path = crate::coordinator::Coordinator::agent_notes_dir(agent)
.join("hyperhive-status");
let result = if text.trim().is_empty() {
std::fs::remove_file(&path).or_else(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
Ok(())
} else {
Err(e)
}
})
} else {
std::fs::write(&path, format!("{}\n", text.trim()))
};
match result {
Ok(()) => {
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
hive_sh4re::Response::Ok
}
Err(e) => {
hive_sh4re::Response::Err {
message: format!("set_status write failed: {e}"),
}
}
}
}
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 role = if target == hive_sh4re::MANAGER_AGENT {
"manager"
} else {
"agent"
}
.to_owned();
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
hive_sh4re::Response::AgentMeta {
name: target.to_owned(),
role,
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
hive_sh4re::Request::CancelLooseEnd { kind, id } => {
crate::questions::handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
|message| hive_sh4re::Response::Err { message },
|()| hive_sh4re::Response::Ok,
)
}
hive_sh4re::Request::AckTurn => match broker.ack_turn(agent) {
Ok(_n) => hive_sh4re::Response::Ok,
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::RequeueInflight => match broker.requeue_inflight(agent) {
Ok(n) => {
if n > 0 {
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
}
hive_sh4re::Response::Ok
}
Err(e) => hive_sh4re::Response::Err {
message: format!("{e:#}"),
},
},
hive_sh4re::Request::GetHostJournal { unit, container, lines, priority, grep, since, until } => {
dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await
}
// Not a shared variant.
_ => return None,
})
}
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 { .. } => match crate::loose_ends::for_agent(coord, agent) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
@ -230,89 +328,6 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
},
}
}
AgentRequest::SetStatus { text } => {
// Cap length + reject multi-line so a confused caller
// can't dump a multi-paragraph session report into the
// dashboard chip.
if let Err(message) = crate::limits::check_status_text(text) {
return AgentResponse::Err { message };
}
let path = crate::coordinator::Coordinator::agent_notes_dir(agent)
.join("hyperhive-status");
let result = if text.trim().is_empty() {
// Empty = clear: remove the file (ignore missing).
std::fs::remove_file(&path)
.or_else(|e| if e.kind() == std::io::ErrorKind::NotFound {
Ok(())
} else {
Err(e)
})
} else {
std::fs::write(&path, format!("{}\n", text.trim()))
};
match result {
Ok(()) => {
// Kick a container rescan so the dashboard updates live.
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
AgentResponse::Ok
}
Err(e) => AgentResponse::Err { message: format!("set_status write failed: {e}") },
}
}
AgentRequest::GetAgentMeta { name } => {
let target = name.as_deref().unwrap_or(agent);
// Gate status on the target's running state so a stopped
// container's stale on-disk status doesn't leak through.
// Also surface `running` itself so callers can tell
// (e.g. "iris is down" vs "iris has no status set").
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let role = if target == hive_sh4re::MANAGER_AGENT {
"manager"
} else {
"agent"
}
.to_owned();
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
AgentResponse::AgentMeta {
name: target.to_owned(),
role,
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
AgentRequest::CancelLooseEnd { kind, id } => crate::questions::handle_cancel_loose_end(
coord, agent, *kind, *id,
)
.map_or_else(
|message| AgentResponse::Err { message },
|()| AgentResponse::Ok,
),
AgentRequest::AckTurn => match broker.ack_turn(agent) {
Ok(_n) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
AgentRequest::RequeueInflight => match broker.requeue_inflight(agent) {
Ok(n) => {
if n > 0 {
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
}
AgentResponse::Ok
}
Err(e) => AgentResponse::Err {
message: format!("{e:#}"),
},
},
AgentRequest::GetHostJournal { unit, container, lines, priority, grep, since, until } => {
dispatch_host_journal(agent, unit, container, lines, priority, grep, since, until).await
}
// Manager-only variants are not valid on the agent socket.
_ => AgentResponse::Err {
message: "request not supported on agent socket".to_owned(),
@ -394,7 +409,7 @@ pub async fn dispatch_host_journal(
/// Fan out one message to each recipient in `targets`. Skips the sender
/// itself. Returns a list of `"<agent>: <error>"` strings for any delivery
/// failures (empty = all good).
fn fan_out_send(
pub(crate) fn fan_out_send(
coord: &Arc<Coordinator>,
from: &str,
body: &str,
@ -421,9 +436,8 @@ fn fan_out_send(
/// Common Send handler shared between dispatch arms. Applies the
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / children fan-out
/// (`to == "<children>"`) / unicast through their respective broker calls.
/// Pulled out of `dispatch` to keep that function under the clippy
/// too-many-lines limit; the behaviour is identical to inlining.
fn handle_send(
/// `pub(crate)` so `dispatch_shared` (and via it, `manager_server`) can use it.
pub(crate) fn handle_send(
coord: &Arc<Coordinator>,
agent: &str,
to: &str,

248
hive-c0re/src/manager_server.rs
View file

@ -6,7 +6,7 @@
use std::sync::Arc;
use anyhow::{Context, Result};
use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse, Message};
use hive_sh4re::{MANAGER_AGENT, ManagerRequest, ManagerResponse};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
@ -74,123 +74,13 @@ async fn serve(stream: UnixStream, coord: Arc<Coordinator>) -> Result<()> {
}
}
/// Max long-poll window for manager `Recv`. Same semantics as the
/// sub-agent socket: omitted `wait_seconds` (or `0`) = peek and
/// return immediately, positive value = park up to that many
/// seconds (clamped at MAX).
const MANAGER_RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_secs(180);
/// Same shape + rationale as `agent_server::RECV_BATCH_MAX`. Kept
/// numerically aligned across surfaces so a tool description that
/// quotes the cap stays accurate either way.
const MANAGER_RECV_BATCH_MAX: u32 = 32;
fn manager_recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
match wait_seconds {
Some(s) => std::time::Duration::from_secs(s).min(MANAGER_RECV_LONG_POLL_MAX),
None => std::time::Duration::ZERO,
}
}
#[allow(clippy::too_many_lines)]
async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResponse {
// Delegate all variants shared with the agent socket to the common handler.
if let Some(resp) = crate::agent_server::dispatch_shared(req, MANAGER_AGENT, coord).await {
return resp;
}
match req {
ManagerRequest::Send {
to,
body,
in_reply_to,
} => {
if let Err(message) = crate::limits::check_size("send", body) {
return ManagerResponse::Err { message };
}
if to == "*" {
let errors = coord.broadcast_send(MANAGER_AGENT, body);
if errors.is_empty() {
ManagerResponse::Ok
} else {
ManagerResponse::Err {
message: format!("broadcast failed for agents: {}", errors.join(", ")),
}
}
} else {
// Resolve magic-recipient sentinels (currently `<parent>`)
// against topology.json; no-op for ordinary names. The
// manager has no parent in topology, so `<parent>`
// resolves to OPERATOR_RECIPIENT — the "no parent → tell
// the operator" fallback. See `docs/conventions.md::
// Recipient sentinels`.
let resolved = crate::topology::resolve_recipient(MANAGER_AGENT, to);
match coord.broker.send(&Message {
from: MANAGER_AGENT.to_owned(),
to: resolved,
body: body.clone(),
in_reply_to: *in_reply_to,
}) {
Ok(()) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
}
ManagerRequest::Wake { from, body } => match coord.broker.send(&Message {
from: from.clone(),
to: MANAGER_AGENT.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
},
ManagerRequest::OperatorMsg { body } => match coord.broker.send(&Message {
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
to: MANAGER_AGENT.to_owned(),
body: body.clone(),
in_reply_to: None,
}) {
Ok(()) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
},
ManagerRequest::Status => match coord.broker.count_pending(MANAGER_AGENT) {
Ok(unread) => ManagerResponse::Status { unread },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
},
ManagerRequest::Recent { limit } => match coord.broker.recent_for(MANAGER_AGENT, *limit) {
Ok(rows) => ManagerResponse::Recent { rows },
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
},
ManagerRequest::Recv { wait_seconds, max } => {
let cap = max.unwrap_or(1).min(MANAGER_RECV_BATCH_MAX) as usize;
match coord
.broker
.recv_blocking_batch(MANAGER_AGENT, manager_recv_timeout(*wait_seconds), cap)
.await
{
Ok(deliveries) => ManagerResponse::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) => ManagerResponse::Err {
message: format!("{e:#}"),
},
}
}
ManagerRequest::RequestInitConfig { name, description } => {
tracing::info!(%name, "manager: request_init_config");
let proposed_dir = crate::coordinator::Coordinator::agent_proposed_dir(name);
@ -375,31 +265,6 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
ManagerRequest::FireScheduleNow { id } => {
handle_fire_schedule_now(coord, hive_sh4re::MANAGER_AGENT, *id).await
}
ManagerRequest::Ask {
question,
options,
multi,
ttl_seconds,
to,
} => crate::questions::handle_ask(
coord,
MANAGER_AGENT,
question,
options,
*multi,
*ttl_seconds,
to.as_deref(),
)
.map_or_else(
|message| ManagerResponse::Err { message },
|id| ManagerResponse::QuestionQueued { id },
),
ManagerRequest::Answer { id, answer } => {
crate::questions::handle_answer(coord, MANAGER_AGENT, *id, answer).map_or_else(
|message| ManagerResponse::Err { message },
|()| ManagerResponse::Ok,
)
}
ManagerRequest::GetLogs { agent, lines } => {
let n = lines.unwrap_or(50);
// `journalctl -M` wants the *machine* name, not the
@ -439,20 +304,6 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
},
}
}
ManagerRequest::Remind {
message,
timing,
file_path,
} => match crate::agent_server::store_remind(
coord,
MANAGER_AGENT,
message,
timing,
file_path.as_deref(),
) {
Ok(()) => ManagerResponse::Ok,
Err(message) => ManagerResponse::Err { message },
},
ManagerRequest::RequestApplyCommit {
agent,
commit_ref,
@ -500,94 +351,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc<Coordinator>) -> ManagerResp
},
}
}
ManagerRequest::SetStatus { text } => {
// Cap length + reject multi-line so a confused caller
// can't dump a multi-paragraph session report into the
// dashboard chip.
if let Err(message) = crate::limits::check_status_text(text) {
return ManagerResponse::Err { message };
}
let path = Coordinator::agent_notes_dir(MANAGER_AGENT).join("hyperhive-status");
let result = if text.trim().is_empty() {
std::fs::remove_file(&path).or_else(|e| {
if e.kind() == std::io::ErrorKind::NotFound {
Ok(())
} else {
Err(e)
}
})
} else {
std::fs::write(&path, format!("{}\n", text.trim()))
};
match result {
Ok(()) => {
let coord2 = Arc::clone(coord);
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("set_status write failed: {e}"),
},
}
}
ManagerRequest::GetAgentMeta { name } => {
let target = name.as_deref().unwrap_or(MANAGER_AGENT);
// Gate status on the target's running state so a stopped
// container's stale on-disk status doesn't leak through.
// Also surface `running` itself so callers can tell
// (e.g. "iris is down" vs "iris has no status set").
let (status_text, status_set_at, running) =
crate::container_view::read_agent_status_live(target).await;
let role = if target == MANAGER_AGENT {
"manager"
} else {
"agent"
}
.to_owned();
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
ManagerResponse::AgentMeta {
name: target.to_owned(),
role,
running,
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
status_text,
status_set_at,
hive_name,
swarm_name,
}
}
ManagerRequest::CancelLooseEnd { kind, id } => {
crate::questions::handle_cancel_loose_end(coord, MANAGER_AGENT, *kind, *id).map_or_else(
|message| ManagerResponse::Err { message },
|()| ManagerResponse::Ok,
)
}
ManagerRequest::AckTurn => match coord.broker.ack_turn(MANAGER_AGENT) {
Ok(_n) => ManagerResponse::Ok,
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
_ => ManagerResponse::Err {
message: "request not handled on manager socket".to_owned(),
},
ManagerRequest::RequeueInflight => match coord.broker.requeue_inflight(MANAGER_AGENT) {
Ok(n) => {
if n > 0 {
tracing::info!(agent = %MANAGER_AGENT, requeued = %n, "requeued in-flight messages");
}
ManagerResponse::Ok
}
Err(e) => ManagerResponse::Err {
message: format!("{e:#}"),
},
},
// GetHostJournal is an agent-socket-only capability-gated variant.
// The manager can use the existing GetLogs tool for per-container
// logs. Route to the agent_server handler for consistency.
ManagerRequest::GetHostJournal { unit, container, lines, priority, grep, since, until } => {
crate::agent_server::dispatch_host_journal(
MANAGER_AGENT, unit, container, lines, priority, grep, since, until,
)
.await
}
}
}