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

address review: drop backwards-compat request/response aliases, use canonical names

Browse source
This commit is contained in:
damocles 2026-07-19 15:12:14 +02:00 • committed by mara
commit 144912f8e0
14 changed files with 183 additions and 229 deletions
Download patch file Download diff file Expand all files Collapse all files

168
hive-c0re/src/socket_server/mod.rs
View file

@ -13,7 +13,7 @@ use std::path::{Path, PathBuf};
use std::sync::Arc;
use anyhow::{Context, Result};
use hive_agent_sock::{AgentRequest, AgentResponse};
use hive_agent_sock::{Request, Response};
use hive_sh4re::{MANAGER_AGENT, Message};
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
use tokio::net::{UnixListener, UnixStream};
@ -146,9 +146,9 @@ async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Re
if n == 0 {
return Ok(());
}
let resp = match serde_json::from_str::<AgentRequest>(line.trim()) {
let resp = match serde_json::from_str::<Request>(line.trim()) {
Ok(req) => dispatch(&req, &agent, &coord).await,
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("parse error: {e}"),
},
};
@ -557,37 +557,37 @@ fn handle_requeue_inflight(coord: &Arc<Coordinator>, agent: &str) -> hive_agent_
/// queries require the `QueryAgentState` capability; hive-wide orchestration
/// verbs (schedules / meta-inputs) require the matching tool-group (the
/// grantable capability).
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
async fn dispatch(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Response {
if let Some(resp) = dispatch_shared(req, agent, coord).await {
return resp;
}
match req {
// Lifecycle + config: caller must be an ancestor of the target
// (a parent owns its whole subtree; the root covers every agent).
AgentRequest::Start { name } => handle_start(coord, agent, name).await,
AgentRequest::Restart { name } => handle_restart(coord, agent, name).await,
AgentRequest::Kill { name } => handle_kill(coord, agent, name).await,
AgentRequest::Update { name } => handle_update(coord, agent, name),
AgentRequest::ListDescendants => handle_list_descendants(agent).await,
AgentRequest::RequestInitConfig { name, description } => {
Request::Start { name } => handle_start(coord, agent, name).await,
Request::Restart { name } => handle_restart(coord, agent, name).await,
Request::Kill { name } => handle_kill(coord, agent, name).await,
Request::Update { name } => handle_update(coord, agent, name),
Request::ListDescendants => handle_list_descendants(agent).await,
Request::RequestInitConfig { name, description } => {
handle_request_init_config(coord, agent, name, description.clone())
}
// Agent-state queries: own subtree is free; other agents + the
// hive-wide `"*"` sweep require `QueryAgentState`.
AgentRequest::GetLooseEnds { agent: target } => {
Request::GetLooseEnds { agent: target } => {
handle_get_loose_ends(coord, agent, target.as_deref())
}
AgentRequest::CountPendingReminders { agent: target } => {
Request::CountPendingReminders { agent: target } => {
handle_count_pending_reminders(coord, agent, target.as_deref())
}
AgentRequest::ReminderRollup {
Request::ReminderRollup {
since_secs,
agent: target,
} => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
// Todos (loose-ends v2): in-container subsystems push/clear
// their own; the agent lists / marks its own done. Scoped to the
// calling agent (the socket identity) — no cross-agent access.
AgentRequest::UpsertTodo {
Request::UpsertTodo {
subsystem,
key,
summary,
@ -600,15 +600,13 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
summary,
source.as_deref(),
),
AgentRequest::ClearTodo {
Request::ClearTodo {
subsystem,
key,
all,
} => handle_clear_todo(coord, agent, subsystem, key.as_deref(), *all),
AgentRequest::ListTodos { subsystem } => {
handle_list_todos(coord, agent, subsystem.as_deref())
}
AgentRequest::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id),
Request::ListTodos { subsystem } => handle_list_todos(coord, agent, subsystem.as_deref()),
Request::MarkTodoDone { id } => handle_mark_todo_done(coord, agent, *id),
// Orchestration / diagnostics verbs — gated per-verb on tool-group
// membership or topology (see `dispatch_orchestration`).
_ => dispatch_orchestration(req, agent, coord).await,
@ -621,13 +619,9 @@ async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) ->
/// (`scheduling` / `approvals`), or `is_descendant_of` for `get_logs`
/// (a parent reads its subtree's logs). Any other variant is a host-admin /
/// unknown request invalid on either socket.
async fn dispatch_orchestration(
req: &AgentRequest,
agent: &str,
coord: &Arc<Coordinator>,
) -> AgentResponse {
async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Response {
match req {
AgentRequest::RequestUpdateMetaInputs {
Request::RequestUpdateMetaInputs {
inputs,
description,
} => {
@ -636,19 +630,19 @@ async fn dispatch_orchestration(
}
handle_request_update_meta_inputs(coord, agent, inputs, description.as_deref())
}
AgentRequest::RequestSchedulePrompt(payload) => {
Request::RequestSchedulePrompt(payload) => {
if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") {
return err;
}
handle_request_schedule_prompt(coord, agent, payload)
}
AgentRequest::CancelSchedule { id, targets } => {
Request::CancelSchedule { id, targets } => {
if let Some(err) = require_group(agent, "scheduling", "cancel a schedule") {
return err;
}
handle_cancel_schedule(coord, agent, *id, targets.as_deref())
}
AgentRequest::EditSchedule {
Request::EditSchedule {
id,
body,
description,
@ -674,19 +668,19 @@ async fn dispatch_orchestration(
},
)
}
AgentRequest::ListSchedules => {
Request::ListSchedules => {
if let Some(err) = require_group(agent, "scheduling", "list schedules") {
return err;
}
handle_list_schedules(coord)
}
AgentRequest::FireScheduleNow { id } => {
Request::FireScheduleNow { id } => {
if let Some(err) = require_group(agent, "scheduling", "fire a schedule") {
return err;
}
handle_fire_schedule_now(coord, agent, *id).await
}
AgentRequest::GetLogs {
Request::GetLogs {
agent: target,
lines,
} => {
@ -696,7 +690,7 @@ async fn dispatch_orchestration(
handle_get_logs(target, *lines).await
}
// Host-admin-only / unknown variants: never valid on either socket.
_ => AgentResponse::Err {
_ => Response::Err {
message: "request not handled on this socket".to_owned(),
},
}
@ -708,11 +702,11 @@ async fn dispatch_orchestration(
/// a consequence, with no positional/hardcoded privilege. Returns `Some(Err)`
/// to short-circuit the dispatch arm when it isn't, `None` when authorised.
/// `action` is the verb phrase for the message (e.g. `"start"`).
fn require_descendant(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
fn require_descendant(agent: &str, target: &str, action: &str) -> Option<Response> {
if crate::topology::is_descendant_of(target, agent) {
None
} else {
Some(AgentResponse::Err {
Some(Response::Err {
message: format!(
"agent `{agent}` cannot {action} `{target}`: \
not in its subtree (topology)"
@ -727,14 +721,14 @@ fn require_descendant(agent: &str, target: &str, action: &str) -> Option<AgentRe
/// capability — granting it to an orchestrator (e.g. the root) authorises
/// these verbs without any positional/hardcoded privilege. `action` is the
/// verb phrase for the message.
fn require_group(agent: &str, group: &str, action: &str) -> Option<AgentResponse> {
fn require_group(agent: &str, group: &str, action: &str) -> Option<Response> {
if crate::tool_groups::groups_for(agent)
.iter()
.any(|g| g == group)
{
None
} else {
Some(AgentResponse::Err {
Some(Response::Err {
message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"),
})
}
@ -753,9 +747,9 @@ fn require_group(agent: &str, group: &str, action: &str) -> Option<AgentResponse
/// could never be a descendant): a brand-new name now flows straight to
/// `submit_init_config`, which builds filesystem paths from it, so validate
/// before that.
fn require_new_child(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
fn require_new_child(agent: &str, target: &str, action: &str) -> Option<Response> {
if let Some(reason) = crate::dashboard::validate_agent_name(target) {
return Some(AgentResponse::Err {
return Some(Response::Err {
message: format!("agent `{agent}` cannot {action} `{target}`: {reason}"),
});
}
@ -771,7 +765,7 @@ fn require_new_child(agent: &str, target: &str, action: &str) -> Option<AgentRes
if crate::topology::is_descendant_of(target, agent) {
None
} else {
Some(AgentResponse::Err {
Some(Response::Err {
message: format!(
"agent `{agent}` cannot {action} `{target}`: it already exists \
outside its subtree in the topology tree"
@ -784,14 +778,10 @@ fn require_new_child(agent: &str, target: &str, action: &str) -> Option<AgentRes
/// descendant resolve freely (a parent sees its subtree, the root sees all);
/// any other named agent needs `QueryAgentState`; `"*"` is a hive-wide sweep
/// gated on `QueryAgentState`.
fn handle_get_loose_ends(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str, target: Option<&str>) -> Response {
let result = if target == Some("*") {
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::QueryAgentState) {
return AgentResponse::Err {
return Response::Err {
message: "query_agent_state capability required for hive-wide loose ends"
.to_owned(),
};
@ -800,12 +790,12 @@ fn handle_get_loose_ends(
} else {
match resolve_agent_state_target(agent, target) {
Ok(name) => crate::loose_ends::for_agent(coord, name),
Err(message) => return AgentResponse::Err { message },
Err(message) => return Response::Err { message },
}
};
match result {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
Ok(loose_ends) => Response::LooseEnds { loose_ends },
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -821,7 +811,7 @@ fn handle_upsert_todo(
key: Option<&str>,
summary: &str,
source: Option<&str>,
) -> AgentResponse {
) -> Response {
match coord.todos.upsert(agent, subsystem, key, summary, source) {
Ok((_, changed)) => {
if changed {
@ -832,9 +822,9 @@ fn handle_upsert_todo(
in_reply_to: None,
});
}
AgentResponse::Ok
Response::Ok
}
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -848,17 +838,17 @@ fn handle_clear_todo(
subsystem: &str,
key: Option<&str>,
all: bool,
) -> AgentResponse {
) -> Response {
let result = if all {
coord.todos.clear_subsystem(agent, subsystem)
} else {
coord.todos.clear(agent, subsystem, key)
};
match result {
Ok(count) => AgentResponse::Acked {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -866,14 +856,10 @@ fn handle_clear_todo(
/// `ListTodos` — enumerate this agent's todos (optionally one subsystem's)
/// as `LooseEnd::Todo` rows, so a producer can reconcile its own set.
fn handle_list_todos(
coord: &Arc<Coordinator>,
agent: &str,
subsystem: Option<&str>,
) -> AgentResponse {
fn handle_list_todos(coord: &Arc<Coordinator>, agent: &str, subsystem: Option<&str>) -> Response {
match crate::loose_ends::todos_for(coord, agent, subsystem) {
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
Err(e) => AgentResponse::Err {
Ok(loose_ends) => Response::LooseEnds { loose_ends },
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -881,12 +867,12 @@ fn handle_list_todos(
/// `MarkTodoDone` — the agent clears one of its own todos by id (scoped to
/// the agent, so it can't touch another agent's).
fn handle_mark_todo_done(coord: &Arc<Coordinator>, agent: &str, id: i64) -> AgentResponse {
fn handle_mark_todo_done(coord: &Arc<Coordinator>, agent: &str, id: i64) -> Response {
match coord.todos.mark_done(agent, id) {
Ok(count) => AgentResponse::Acked {
Ok(count) => Response::Acked {
count: u64::try_from(count).unwrap_or(0),
},
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -898,15 +884,15 @@ fn handle_count_pending_reminders(
coord: &Arc<Coordinator>,
agent: &str,
target: Option<&str>,
) -> AgentResponse {
) -> Response {
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 {
Ok(count) => Response::PendingRemindersCount { count },
Err(e) => Response::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
Err(message) => Response::Err { message },
}
}
@ -918,15 +904,15 @@ fn handle_reminder_rollup(
agent: &str,
target: Option<&str>,
since_secs: u64,
) -> AgentResponse {
) -> Response {
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 {
Ok(stats) => Response::ReminderRollup(stats),
Err(e) => Response::Err {
message: format!("{e:#}"),
},
},
Err(message) => AgentResponse::Err { message },
Err(message) => Response::Err { message },
}
}
@ -949,7 +935,7 @@ pub struct HostJournalArgs<'a> {
///
/// 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 {
pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Response {
let HostJournalArgs {
unit,
container,
@ -960,7 +946,7 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag
until,
} = args;
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) {
return AgentResponse::Err {
return Response::Err {
message: "agent does not have the read_host_journal capability".to_owned(),
};
}
@ -988,9 +974,9 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag
{
Ok((stdout, stderr)) => {
let content = if stdout.is_empty() { stderr } else { stdout };
AgentResponse::HostJournal { content }
Response::HostJournal { content }
}
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("journal read: {e:#}"),
},
};
@ -1032,9 +1018,9 @@ pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> Ag
let stderr = String::from_utf8_lossy(&out.stderr);
format!("journalctl exited {}: {stderr}", out.status)
};
AgentResponse::HostJournal { content }
Response::HostJournal { content }
}
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("journalctl spawn failed: {e:#}"),
},
}
@ -1077,16 +1063,16 @@ pub(crate) fn handle_send(
to: &str,
body: &str,
in_reply_to: Option<i64>,
) -> AgentResponse {
) -> Response {
if let Err(message) = crate::limits::check_size("send", body) {
return AgentResponse::Err { message };
return Response::Err { message };
}
if to == "*" {
let errors = coord.broadcast_send(agent, body);
return if errors.is_empty() {
AgentResponse::Ok
Response::Ok
} else {
AgentResponse::Err {
Response::Err {
message: format!("broadcast failed for agents: {}", errors.join(", ")),
}
};
@ -1099,9 +1085,9 @@ pub(crate) fn handle_send(
let children = crate::topology::children_of(agent);
let errors = fan_out_send(coord, agent, body, in_reply_to, &children);
return if errors.is_empty() {
AgentResponse::Ok
Response::Ok
} else {
AgentResponse::Err {
Response::Err {
message: format!("children fan-out failed for agents: {}", errors.join(", ")),
}
};
@ -1118,7 +1104,7 @@ pub(crate) fn handle_send(
// Cross-hive messaging (`name@hive` qualified names) is not routed
// through the broker — use the Matrix MCP tools for that instead.
if resolved.contains('@') {
return AgentResponse::Err {
return Response::Err {
message: format!(
"send failed: cross-hive recipient `{resolved}` is not supported \
via the broker — use Matrix MCP tools for cross-hive messaging"
@ -1128,7 +1114,7 @@ pub(crate) fn handle_send(
if resolved != hive_sh4re::OPERATOR_RECIPIENT {
let state_root = crate::paths::agent_state_dir(&resolved);
if !state_root.exists() {
return AgentResponse::Err {
return Response::Err {
message: format!(
"send failed: unknown recipient `{resolved}` \
(no agent with that name exists on this hive)"
@ -1142,8 +1128,8 @@ pub(crate) fn handle_send(
body: body.to_owned(),
in_reply_to,
}) {
Ok(()) => AgentResponse::Ok,
Err(e) => AgentResponse::Err {
Ok(()) => Response::Ok,
Err(e) => Response::Err {
message: format!("{e:#}"),
},
}
@ -1152,7 +1138,7 @@ pub(crate) fn handle_send(
/// `GetLogs` — read a child container's journal via hive-priv (the
/// `-M` read needs root). `journalctl -M` wants the `h-<name>` machine
/// name, which `container_name` derives.
async fn handle_get_logs(agent: &str, lines: Option<u32>) -> AgentResponse {
async fn handle_get_logs(agent: &str, lines: Option<u32>) -> Response {
let n = lines.unwrap_or(50);
let machine = crate::lifecycle::container_name(agent);
tracing::info!(%agent, %machine, %n, "manager: get_logs");
@ -1167,9 +1153,9 @@ async fn handle_get_logs(agent: &str, lines: Option<u32>) -> AgentResponse {
{
Ok((stdout, stderr)) => {
let content = if stdout.is_empty() { stderr } else { stdout };
AgentResponse::Logs { content }
Response::Logs { content }
}
Err(e) => AgentResponse::Err {
Err(e) => Response::Err {
message: format!("get_logs: {e:#}"),
},
}