diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index e50f6d01..6cf8cb1b 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -1,7 +1,7 @@ //! Embedded MCP server. Claude Code (running inside the agent container) //! connects to this over streamable-HTTP via `--mcp-config` (the long-lived //! `hive-mcp-http` daemon); tool calls land here and are translated to -//! `AgentRequest::*` / `ManagerRequest::*` against hyperhive's own +//! `Request::*` against hyperhive's own //! per-container unix socket at `/run/hive/mcp.sock`. //! //! Two protocols, two surfaces: @@ -99,11 +99,9 @@ where /// Unified MCP tool surface for both sub-agent and manager roles. /// -/// `AgentRequest = ManagerRequest = Request` and `AgentResponse = -/// ManagerResponse = Response` are type aliases in hive-sh4re, so a single +/// Both sockets speak the same `Request` / `Response` wire, so a single /// `dispatch` call covers both sockets — the only real difference is which /// socket path is used and which tools the flavor enables. -/// #[derive(Debug, Clone)] pub struct AgentServer { socket: PathBuf, @@ -119,8 +117,8 @@ impl AgentServer { /// `Response` plus the retry count so tool handlers can annotate their /// result (see `annotate_retries`). /// - /// `AgentRequest` / `ManagerRequest` / `Request` are all the same type - /// (hive-sh4re type aliases), so this single method covers both sockets. + /// Both sockets speak the same `Request` type, so this single method + /// covers both. async fn dispatch( &self, req: hive_agent_sock::Request, diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index 5ad53ff1..e3ef2093 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -21,9 +21,6 @@ fn default_true() -> bool { /// Unified request enum for both agent and manager sockets. The agent's /// identity is the socket it arrived on. Privileged variants are marked /// `*(privileged)*` — an agent socket returns `Err` for them server-side. -/// -/// `AgentRequest` and `ManagerRequest` are type aliases for this enum; -/// existing callers continue to compile unchanged. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "cmd", rename_all = "snake_case")] pub enum Request { @@ -308,16 +305,8 @@ pub enum Request { }, } -/// Backwards-compatible aliases. Both sockets now speak the unified `Request` -/// / `Response` wire; the server-side privilege gate rejects privileged -/// variants on agent sockets with `Err { message: "privileged variant..." }`. -pub type AgentRequest = Request; -pub type ManagerRequest = Request; - /// Unified response enum for both agent and manager sockets. Privileged /// variants (`Logs`, `Schedules`) are never returned on agent sockets. -/// -/// `AgentResponse` and `ManagerResponse` are type aliases for this enum. #[derive(Debug, Clone, Serialize, Deserialize)] #[serde(tag = "kind", rename_all = "snake_case")] pub enum Response { @@ -408,7 +397,3 @@ pub enum Response { /// the agent's next start. GracefulStop, } - -/// Backwards-compatible response aliases. -pub type AgentResponse = Response; -pub type ManagerResponse = Response; diff --git a/hive-agent-wake/src/main.rs b/hive-agent-wake/src/main.rs index 2cf8b012..d21cfca9 100644 --- a/hive-agent-wake/src/main.rs +++ b/hive-agent-wake/src/main.rs @@ -13,7 +13,7 @@ use std::path::PathBuf; use anyhow::Result; use clap::Parser; -use hive_agent_sock::{AgentRequest, AgentResponse}; +use hive_agent_sock::{Request, Response}; /// Per-agent MCP socket, bind-mounted from the host into every container. const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock"; @@ -50,17 +50,17 @@ async fn main() -> Result<()> { } else { cli.body }; - let resp: AgentResponse = client::request( + let resp: Response = client::request( &cli.socket, - &AgentRequest::Wake { + &Request::Wake { from: cli.from, body, }, ) .await?; match resp { - AgentResponse::Ok => Ok(()), - AgentResponse::Err { message } => anyhow::bail!("wake: {message}"), + Response::Ok => Ok(()), + Response::Err { message } => anyhow::bail!("wake: {message}"), other => anyhow::bail!("wake: unexpected response {other:?}"), } } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index bad8f4b3..7e0d98c4 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -42,7 +42,7 @@ use crate::login::LoginState; use crate::turn_stats::TurnStats; use anyhow::Result; use clap::Parser; -use hive_agent_sock::{AgentRequest, AgentResponse}; +use hive_agent_sock::{Request, Response}; use hive_sh4re::{HelperEvent, SYSTEM_SENDER}; #[derive(Parser)] @@ -191,9 +191,8 @@ fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { // ---------- surface trait ---------- -/// What a `Recv` long-poll returned. Decoupled from the per-role -/// Response enum so `serve_loop` can pattern-match without seeing -/// either `AgentResponse` or `ManagerResponse` directly. +/// What a `Recv` long-poll returned. Decoupled from the `Response` +/// enum so `serve_loop` can pattern-match without seeing it directly. enum RecvOutcome { /// Long-poll returned at least one message; first one is detached. Message(hive_sh4re::DeliveredMessage), @@ -213,7 +212,7 @@ enum RecvOutcome { /// Wire surface abstraction. `AgentSurface` is the only impl — the trait /// exists to keep the turn loop generic and testable. Every function that /// talks to the broker goes through this so there are zero hard-coded -/// `AgentRequest` / `AgentResponse` references in the turn loop itself. +/// `Request` / `Response` references in the turn loop itself. trait Surface { /// Ack the in-flight turn. Logs warnings on transport/broker /// errors but never propagates — turn loop continues either way. @@ -252,17 +251,17 @@ trait Surface { // ---------- AgentSurface ---------- /// Zero-sized type tag for the agent wire surface. -/// Talks `AgentRequest` / `AgentResponse`. +/// Talks `Request` / `Response`. struct AgentSurface; /// Issue an `Ok`-expecting fire-and-forget broker request, logging any /// rejection / unexpected response / transport error under `label`. Shared by /// the `Surface` methods that don't need the reply (`ack_turn`, /// `requeue_inflight`, `graceful_stop_complete`). -async fn fire_and_forget(socket: &Path, req: AgentRequest, label: &str) { - match client::request::<_, AgentResponse>(socket, &req).await { - Ok(AgentResponse::Ok) => {} - Ok(AgentResponse::Err { message }) => { +async fn fire_and_forget(socket: &Path, req: Request, label: &str) { + match client::request::<_, Response>(socket, &req).await { + Ok(Response::Ok) => {} + Ok(Response::Err { message }) => { tracing::warn!(%message, "{label} rejected by broker"); } Ok(other) => tracing::warn!(?other, "{label} unexpected response"), @@ -272,55 +271,53 @@ async fn fire_and_forget(socket: &Path, req: AgentRequest, label: &str) { impl Surface for AgentSurface { async fn ack_turn(socket: &Path) { - fire_and_forget(socket, AgentRequest::AckTurn, "ack_turn").await; + fire_and_forget(socket, Request::AckTurn, "ack_turn").await; } async fn requeue_inflight(socket: &Path) { - fire_and_forget(socket, AgentRequest::RequeueInflight, "requeue_inflight").await; + fire_and_forget(socket, Request::RequeueInflight, "requeue_inflight").await; } async fn graceful_stop_complete(socket: &Path) { fire_and_forget( socket, - AgentRequest::GracefulStopComplete, + Request::GracefulStopComplete, "graceful_stop_complete", ) .await; } async fn inbox_unread(socket: &Path) -> u64 { - match client::request::<_, AgentResponse>(socket, &AgentRequest::Status).await { - Ok(AgentResponse::Status { unread }) => unread, + match client::request::<_, Response>(socket, &Request::Status).await { + Ok(Response::Status { unread }) => unread, _ => 0, } } async fn post_turn_counts(socket: &Path) -> (Option, Option) { - let threads = match client::request::<_, AgentResponse>( + let threads = + match client::request::<_, Response>(socket, &Request::GetLooseEnds { agent: None }) + .await + { + Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), + _ => None, + }; + let reminders = match client::request::<_, Response>( socket, - &AgentRequest::GetLooseEnds { agent: None }, + &Request::CountPendingReminders { agent: None }, ) .await { - Ok(AgentResponse::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), - _ => None, - }; - let reminders = match client::request::<_, AgentResponse>( - socket, - &AgentRequest::CountPendingReminders { agent: None }, - ) - .await - { - Ok(AgentResponse::PendingRemindersCount { count }) => Some(count), + Ok(Response::PendingRemindersCount { count }) => Some(count), _ => None, }; (threads, reminders) } async fn send_to_parent(socket: &Path, body: String) { - let res = client::request::<_, AgentResponse>( + let res = client::request::<_, Response>( socket, - &AgentRequest::Send { + &Request::Send { to: hive_sh4re::PARENT_RECIPIENT.into(), body, in_reply_to: None, @@ -333,22 +330,22 @@ impl Surface for AgentSurface { } async fn recv_next(socket: &Path) -> RecvOutcome { - let recv: Result = client::request( + let recv: Result = client::request( socket, - &AgentRequest::Recv { + &Request::Recv { wait_seconds: Some(180), max: None, }, ) .await; match recv { - Ok(AgentResponse::Messages { messages, .. }) if !messages.is_empty() => { + Ok(Response::Messages { messages, .. }) if !messages.is_empty() => { let first = messages.into_iter().next().expect("checked non-empty"); RecvOutcome::Message(first) } - Ok(AgentResponse::Messages { .. }) => RecvOutcome::Empty, - Ok(AgentResponse::GracefulStop) => RecvOutcome::GracefulStop, - Ok(AgentResponse::Err { message }) => { + Ok(Response::Messages { .. }) => RecvOutcome::Empty, + Ok(Response::GracefulStop) => RecvOutcome::GracefulStop, + Ok(Response::Err { message }) => { tracing::warn!(%message, "recv error"); RecvOutcome::TransportError } diff --git a/hive-bash-mcp/src/runner.rs b/hive-bash-mcp/src/runner.rs index adcab4c0..1db267f1 100644 --- a/hive-bash-mcp/src/runner.rs +++ b/hive-bash-mcp/src/runner.rs @@ -760,7 +760,7 @@ pub(crate) async fn send_wake( ); } } - let req = hive_agent_sock::AgentRequest::Wake { + let req = hive_agent_sock::Request::Wake { from: format!("bash-task-{id}"), body, }; diff --git a/hive-c0re/src/agent_config/limits.rs b/hive-c0re/src/agent_config/limits.rs index ad981f63..723fc58f 100644 --- a/hive-c0re/src/agent_config/limits.rs +++ b/hive-c0re/src/agent_config/limits.rs @@ -23,7 +23,7 @@ pub const MESSAGE_MAX_BYTES: usize = 4096; /// Validate that `body` fits under [`MESSAGE_MAX_BYTES`]. Returns a /// caller-ready error string (caller wraps in -/// `AgentResponse::Err`/`ManagerResponse::Err`) on failure. +/// `Response::Err`) on failure. /// /// `label` shows up in the error message verbatim — pass a short /// noun like `"send"`, `"question"`, `"broadcast"` so the model can diff --git a/hive-c0re/src/dashboard/lifecycle_ops.rs b/hive-c0re/src/dashboard/lifecycle_ops.rs index 7960fe49..4f2bdb9f 100644 --- a/hive-c0re/src/dashboard/lifecycle_ops.rs +++ b/hive-c0re/src/dashboard/lifecycle_ops.rs @@ -77,7 +77,7 @@ pub(super) async fn post_kill( // host-side approval queue without the manager up, and // operator-driven meta-input updates work from the dashboard // either way. The MCP-surface self-kill guard in - // `socket_server.rs::ManagerRequest::Kill` stays in place: a + // `socket_server.rs::Request::Kill` stays in place: a // manager calling Kill on its own container is self-suicide // mid-call, not a legitimate operator action. submit::stop( diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index b4803b37..72a9b0ee 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -1,8 +1,8 @@ //! Loose-ends aggregator. Walks the `approvals` + `operator_questions` //! tables once per call and assembles a `Vec` for either -//! a single agent (`for_agent`) or the whole hive (`hive_wide`). Both -//! `AgentRequest::GetLooseEnds` and `ManagerRequest::GetLooseEnds` -//! land here so the routing logic + age-seconds derivation stay in +//! a single agent (`for_agent`) or the whole hive (`hive_wide`). +//! `Request::GetLooseEnds` from either the agent or manager socket +//! lands here so the routing logic + age-seconds derivation stay in //! one place. //! //! Call frequency is low (an agent doing self-introspection between diff --git a/hive-c0re/src/socket_server/config_approvals.rs b/hive-c0re/src/socket_server/config_approvals.rs index 572596fe..3b35f8be 100644 --- a/hive-c0re/src/socket_server/config_approvals.rs +++ b/hive-c0re/src/socket_server/config_approvals.rs @@ -9,7 +9,7 @@ use std::sync::Arc; -use hive_agent_sock::AgentResponse; +use hive_agent_sock::Response; use super::require_new_child; use crate::coordinator::Coordinator; @@ -24,14 +24,14 @@ pub(super) fn handle_request_init_config( agent: &str, name: &str, description: Option, -) -> AgentResponse { +) -> Response { if let Some(err) = require_new_child(agent, name, "request_init_config for") { return err; } tracing::info!(%agent, %name, "request_init_config"); match submit_init_config(coord, name, Some(agent), description) { - Ok(_id) => AgentResponse::Ok, - Err(e) => AgentResponse::Err { + Ok(_id) => Response::Ok, + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -45,7 +45,7 @@ pub(super) fn handle_request_update_meta_inputs( requester: &str, inputs: &[String], description: Option<&str>, -) -> AgentResponse { +) -> Response { let label = if inputs.is_empty() { "all inputs".to_string() } else { @@ -67,7 +67,7 @@ pub(super) fn handle_request_update_meta_inputs( { Ok(id) => id, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("queue update_meta_inputs approval: {e:#}"), }; } @@ -81,7 +81,7 @@ pub(super) fn handle_request_update_meta_inputs( description: description.map(str::to_owned), pr_number: None, }); - AgentResponse::Ok + Response::Ok } /// Submit-time half of the PR-merge flow: fetch the PR head sha from the diff --git a/hive-c0re/src/socket_server/lifecycle_handlers.rs b/hive-c0re/src/socket_server/lifecycle_handlers.rs index 892536ca..ae57b3bd 100644 --- a/hive-c0re/src/socket_server/lifecycle_handlers.rs +++ b/hive-c0re/src/socket_server/lifecycle_handlers.rs @@ -5,18 +5,14 @@ use std::sync::Arc; -use hive_agent_sock::AgentResponse; +use hive_agent_sock::Response; use super::require_descendant; use crate::coordinator::Coordinator; /// `Start` — start a container, kicking its next turn. The caller must be an /// ancestor of `name` in the topology (the root covers every agent). -pub(super) async fn handle_start( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_start(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "start") { return err; } @@ -31,18 +27,14 @@ pub(super) async fn handle_start( format!("agent `{agent}` start tool"), ) .await; - AgentResponse::Ok + Response::Ok } /// `Restart` — enqueue a restart for a container. The caller must be an /// ancestor of `name` in the topology. The infra-container branch is /// orthogonal: it is gated on the `infra_admin` capability and audited, so it /// stays ahead of the topology guard. -pub(super) async fn handle_restart( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_restart(coord: &Arc, agent: &str, name: &str) -> Response { // Infra-container restart: an agent holding the `infra_admin` // capability can restart a hive infrastructure container (hive-ci / // hive-gateway / hive-forge / hive-matrix) by passing its name to the @@ -63,7 +55,7 @@ pub(super) async fn handle_restart( format!("agent `{agent}` restart tool"), ) .await; - AgentResponse::Ok + Response::Ok } /// Restart a hive infrastructure container on behalf of an agent that @@ -75,7 +67,7 @@ async fn handle_restart_infra( coord: &Arc, agent: &str, container: hive_priv_sock::InfraContainer, -) -> AgentResponse { +) -> Response { let name = container.unit_name(); // Record the attempt in the operator-visible privileged-action audit // trail, then emit a live `AuditEntryAdded` so the dashboard audit view @@ -97,7 +89,7 @@ async fn handle_restart_infra( crate::audit_log::AuditOutcome::Err, Some("denied: missing infra_admin capability"), ); - return AgentResponse::Err { + return Response::Err { message: format!( "restarting infra container `{name}` requires the `infra_admin` capability" ), @@ -107,23 +99,19 @@ async fn handle_restart_infra( match crate::priv_client::restart_infra_container(container).await { Ok(()) => { audit(crate::audit_log::AuditOutcome::Ok, None); - AgentResponse::Ok + Response::Ok } Err(e) => { let msg = format!("{e:#}"); audit(crate::audit_log::AuditOutcome::Err, Some(&msg)); - AgentResponse::Err { message: msg } + Response::Err { message: msg } } } } /// `Kill` — kill a container, unregister it, notify the manager. The caller /// must be an ancestor of `name` in the topology. -pub(super) async fn handle_kill( - coord: &Arc, - agent: &str, - name: &str, -) -> AgentResponse { +pub(super) async fn handle_kill(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "kill") { return err; } @@ -144,9 +132,9 @@ pub(super) async fn handle_kill( coord.notify_manager(&hive_sh4re::HelperEvent::Killed { agent: name.to_owned(), }); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("{e:#}"), }, } @@ -154,7 +142,7 @@ pub(super) async fn handle_kill( /// `Update` — enqueue a rebuild for a container. The caller must be an /// ancestor of `name` in the topology. -pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) -> AgentResponse { +pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) -> Response { if let Some(err) = require_descendant(agent, name, "rebuild") { return err; } @@ -165,12 +153,12 @@ pub(super) fn handle_update(coord: &Arc, agent: &str, name: &str) - crate::job_queue::Source::Manual, format!("agent `{agent}` update tool"), ); - AgentResponse::Ok + Response::Ok } /// `ListDescendants` — every topological descendant of `agent` with /// its running/stopped state, parents before children. -pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { +pub(super) async fn handle_list_descendants(agent: &str) -> Response { 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 { @@ -182,7 +170,7 @@ pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { }) .collect(), Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("list containers failed: {e:#}"), }; } @@ -203,5 +191,5 @@ pub(super) async fn handle_list_descendants(agent: &str) -> AgentResponse { hive_sh4re::ContainerInfo { name, running } }) .collect(); - AgentResponse::Containers { containers } + Response::Containers { containers } } diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index 42536319..7b04daf3 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -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) -> Re if n == 0 { return Ok(()); } - let resp = match serde_json::from_str::(line.trim()) { + let resp = match serde_json::from_str::(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, 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) -> AgentResponse { +async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> 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) -> 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) -> /// (`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, -) -> AgentResponse { +async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc) -> 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 { +fn require_descendant(agent: &str, target: &str, action: &str) -> Option { 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 Option { +fn require_group(agent: &str, group: &str, action: &str) -> Option { 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 Option { +fn require_new_child(agent: &str, target: &str, action: &str) -> Option { 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 Option, - agent: &str, - target: Option<&str>, -) -> AgentResponse { +fn handle_get_loose_ends(coord: &Arc, 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, - agent: &str, - subsystem: Option<&str>, -) -> AgentResponse { +fn handle_list_todos(coord: &Arc, 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, agent: &str, id: i64) -> AgentResponse { +fn handle_mark_todo_done(coord: &Arc, 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, 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, -) -> 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-` machine /// name, which `container_name` derives. -async fn handle_get_logs(agent: &str, lines: Option) -> AgentResponse { +async fn handle_get_logs(agent: &str, lines: Option) -> 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) -> 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:#}"), }, } diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs index 89e2f459..e2b7b734 100644 --- a/hive-c0re/src/socket_server/reminders.rs +++ b/hive-c0re/src/socket_server/reminders.rs @@ -5,7 +5,7 @@ use std::sync::Arc; -use hive_agent_sock::AgentResponse; +use hive_agent_sock::Response; use crate::coordinator::Coordinator; @@ -15,10 +15,10 @@ pub(super) fn handle_remind( message: &str, timing: &hive_sh4re::ReminderTiming, file_path: Option<&str>, -) -> AgentResponse { +) -> Response { match store_remind(coord, agent, message, timing, file_path) { - Ok(()) => AgentResponse::Ok, - Err(message) => AgentResponse::Err { message }, + Ok(()) => Response::Ok, + Err(message) => Response::Err { message }, } } diff --git a/hive-c0re/src/socket_server/schedules.rs b/hive-c0re/src/socket_server/schedules.rs index 0e945ac1..9ba19b1b 100644 --- a/hive-c0re/src/socket_server/schedules.rs +++ b/hive-c0re/src/socket_server/schedules.rs @@ -6,17 +6,17 @@ use std::sync::Arc; -use hive_agent_sock::AgentResponse; +use hive_agent_sock::Response; use crate::coordinator::Coordinator; /// `ListSchedules` — snapshot every scheduled prompt onto the wire. -pub(super) fn handle_list_schedules(coord: &Arc) -> AgentResponse { +pub(super) fn handle_list_schedules(coord: &Arc) -> Response { match coord.scheduled_prompts.list() { - Ok(schedules) => AgentResponse::Schedules { + Ok(schedules) => Response::Schedules { schedules: schedules.into_iter().map(schedule_to_wire).collect(), }, - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("list scheduled prompts: {e:#}"), }, } @@ -32,26 +32,26 @@ pub(super) fn handle_request_schedule_prompt( coord: &Arc, requester: &str, payload: &hive_sh4re::SchedulePromptPayload, -) -> AgentResponse { +) -> Response { if payload.targets.is_empty() { - return AgentResponse::Err { + return Response::Err { message: "schedule must have at least one target".into(), }; } if payload.body.trim().is_empty() { - return AgentResponse::Err { + return Response::Err { message: "schedule body must be non-empty".into(), }; } if let Some(0) = payload.interval_seconds { - return AgentResponse::Err { + return Response::Err { message: "interval_seconds must be > 0 (use None for one-shot)".into(), }; } let commit_ref = match serde_json::to_string(payload) { Ok(s) => s, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("encode SchedulePromptPayload: {e:#}"), }; } @@ -66,7 +66,7 @@ pub(super) fn handle_request_schedule_prompt( ) { Ok(id) => id, Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("queue schedule_prompt approval: {e:#}"), }; } @@ -87,7 +87,7 @@ pub(super) fn handle_request_schedule_prompt( description: payload.description.clone(), pr_number: None, }); - AgentResponse::Ok + Response::Ok } /// Cancel a schedule (whole or per-target). Manager-surface @@ -101,22 +101,22 @@ pub(super) fn handle_cancel_schedule( requester: &str, schedule_id: i64, targets: Option<&[String]>, -) -> AgentResponse { +) -> Response { let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot cancel schedule owned by {owner}", owner = schedule.owner @@ -136,9 +136,9 @@ pub(super) fn handle_cancel_schedule( match result { Ok(()) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(message) => AgentResponse::Err { message }, + Err(message) => Response::Err { message }, } } @@ -151,22 +151,22 @@ pub(super) async fn handle_fire_schedule_now( coord: &Arc, requester: &str, schedule_id: i64, -) -> AgentResponse { +) -> Response { let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot fire schedule owned by {owner}", owner = schedule.owner @@ -178,9 +178,9 @@ pub(super) async fn handle_fire_schedule_now( match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await { Ok(_report) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("fire schedule {schedule_id} now: {e:#}"), }, } @@ -217,7 +217,7 @@ pub(super) fn handle_edit_schedule( requester: &str, schedule_id: i64, patch: EditSchedulePatch, -) -> AgentResponse { +) -> Response { let EditSchedulePatch { body, description, @@ -229,18 +229,18 @@ pub(super) fn handle_edit_schedule( let schedule = match coord.scheduled_prompts.get(schedule_id) { Ok(Some(s)) => s, Ok(None) => { - return AgentResponse::Err { + return Response::Err { message: format!("schedule {schedule_id} not found"), }; } Err(e) => { - return AgentResponse::Err { + return Response::Err { message: format!("read schedule {schedule_id}: {e:#}"), }; } }; if !cancel_authorized(requester, &schedule.owner) { - return AgentResponse::Err { + return Response::Err { message: format!( "not authorized: {requester} cannot edit schedule owned by {owner}", owner = schedule.owner @@ -258,9 +258,9 @@ pub(super) fn handle_edit_schedule( match coord.scheduled_prompts.update(schedule_id, patch) { Ok(()) => { coord.emit_schedules_snapshot(); - AgentResponse::Ok + Response::Ok } - Err(e) => AgentResponse::Err { + Err(e) => Response::Err { message: format!("edit schedule {schedule_id}: {e:#}"), }, } diff --git a/hive-matrix-mcp/src/wake.rs b/hive-matrix-mcp/src/wake.rs index 2ebe4299..2ce4e737 100644 --- a/hive-matrix-mcp/src/wake.rs +++ b/hive-matrix-mcp/src/wake.rs @@ -3,7 +3,7 @@ //! //! Same wire shape as `hive-agent::forge_notify`'s wake: a single JSON //! line written to the hyperhive control socket (`/run/hive/mcp.sock` -//! by default) carrying an `AgentRequest::Wake { from, body }`. +//! by default) carrying an `Request::Wake { from, body }`. //! The agent harness's `agent_server` parses it and treats it as a //! `Wake` from the matrix subsystem. //! @@ -19,7 +19,7 @@ use anyhow::{Context, Result}; use tokio::io::AsyncWriteExt; use tokio::net::UnixStream; -/// Send an `AgentRequest::Wake { from: "matrix", body }` to the hyperhive +/// Send an `Request::Wake { from: "matrix", body }` to the hyperhive /// control socket at `socket`. Best-effort: returns Err on any plumbing /// failure; callers log + ignore so a wake delivery hiccup doesn't tear /// down the matrix sync loop.