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

refactor: unify AgentRequest/Response + ManagerRequest/Response into Request/Response (#691)

Browse source
This commit is contained in:
damocles 2026-06-01 09:30:44 +02:00 • committed by mara
commit 15617cef9a
6 changed files with 197 additions and 534 deletions
Download patch file Download diff file Expand all files Collapse all files

19
hive-ag3nt/src/bin/hive.rs
View file

@ -188,14 +188,6 @@ trait Surface {
/// system-prompt block + tool registration goes into the spawned
/// `claude` process.
const FLAVOR: mcp::Flavor;
/// `is_manager` flag passed to `forge_notify::run`. Picks which
/// wire enum (`AgentRequest::Wake` vs `ManagerRequest::Wake`) the
/// poller uses to push notifications into the harness inbox — the
/// per-role broker socket rejects the wrong type. Lifting
/// `Surface` into the lib crate to make `forge_notify::run`
/// generic is deferred to its own issue.
const FORGE_IS_MANAGER: bool;
/// Ack the in-flight turn. Logs warnings on transport/broker
/// errors but never propagates — turn loop continues either way.
fn ack_turn(socket: &Path) -> impl Future<Output = ()>;
@ -248,7 +240,6 @@ struct AgentSurface;
impl Surface for AgentSurface {
const FLAVOR: mcp::Flavor = mcp::Flavor::Agent;
const FORGE_IS_MANAGER: bool = false;
async fn ack_turn(socket: &Path) {
match client::request::<_, AgentResponse>(socket, &AgentRequest::AckTurn).await {
@ -281,7 +272,7 @@ impl Surface for AgentSurface {
async fn post_turn_counts(socket: &Path) -> (Option<u64>, Option<u64>) {
let threads =
match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds).await {
match client::request::<_, AgentResponse>(socket, &AgentRequest::GetLooseEnds { agent: None }).await {
Ok(AgentResponse::LooseEnds { loose_ends }) => {
u64::try_from(loose_ends.len()).ok()
}
@ -289,7 +280,7 @@ impl Surface for AgentSurface {
};
let reminders = match client::request::<_, AgentResponse>(
socket,
&AgentRequest::CountPendingReminders,
&AgentRequest::CountPendingReminders { agent: None },
)
.await
{
@ -386,7 +377,6 @@ struct ManagerSurface;
impl Surface for ManagerSurface {
const FLAVOR: mcp::Flavor = mcp::Flavor::Manager;
const FORGE_IS_MANAGER: bool = true;
async fn ack_turn(socket: &Path) {
match client::request::<_, ManagerResponse>(socket, &ManagerRequest::AckTurn).await {
@ -560,10 +550,7 @@ async fn serve_main<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
for failure in plugins::install_configured(socket).await {
S::send_to_parent(socket, failure).await;
}
tokio::spawn(hive_ag3nt::forge_notify::run(
socket.to_path_buf(),
S::FORGE_IS_MANAGER,
));
tokio::spawn(hive_ag3nt::forge_notify::run(socket.to_path_buf()));
// Log web_ui::serve's error instead of dropping it. A bare
// `tokio::spawn(web_ui::serve(...))` discards the JoinHandle, so
// any Err (e.g. EACCES from `bind_unix` when HIVE_WEB_SOCKET points

29
hive-ag3nt/src/forge_notify.rs
View file

@ -26,10 +26,7 @@ const BODY_TRUNCATE: usize = 500;
/// configured. Otherwise loops forever, polling every
/// `POLL_INTERVAL_SECS` seconds. Errors are never fatal.
///
/// `is_manager`: when true, wakes the inbox via `ManagerRequest::Wake`
/// instead of `AgentRequest::Wake` (the manager socket rejects the agent
/// request type).
pub async fn run(socket: PathBuf, is_manager: bool) {
pub async fn run(socket: PathBuf) {
let forge_url = match std::env::var("HIVE_FORGE_URL") {
Ok(u) if !u.is_empty() => u,
_ => {
@ -122,7 +119,6 @@ pub async fn run(socket: PathBuf, is_manager: bool) {
&forge_url,
&token,
&socket,
is_manager,
keep_subscriptions,
&mut unsubbed_repos,
&own_login,
@ -623,7 +619,6 @@ async fn poll_once(
forge_url: &str,
token: &str,
socket: &Path,
is_manager: bool,
keep_subscriptions: bool,
unsubbed_repos: &mut HashSet<String>,
own_login: &str,
@ -690,23 +685,13 @@ async fn poll_once(
continue;
};
let delivered = if is_manager {
let req = hive_sh4re::ManagerRequest::Wake {
from: "forge".to_owned(),
body,
};
crate::client::request::<_, hive_sh4re::ManagerResponse>(socket, &req)
.await
.map(|_| ())
} else {
let req = hive_sh4re::AgentRequest::Wake {
from: "forge".to_owned(),
body,
};
crate::client::request::<_, hive_sh4re::AgentResponse>(socket, &req)
.await
.map(|_| ())
let req = hive_sh4re::Request::Wake {
from: "forge".to_owned(),
body,
};
let delivered = crate::client::request::<_, hive_sh4re::Response>(socket, &req)
.await
.map(|_| ());
match delivered {
Ok(()) => {
debug!(%id, "forge_notify: delivered");

67
hive-ag3nt/src/mcp.rs
View file

@ -64,60 +64,23 @@ pub enum SocketReply {
},
}
impl From<hive_sh4re::AgentResponse> for SocketReply {
fn from(r: hive_sh4re::AgentResponse) -> Self {
impl From<hive_sh4re::Response> for SocketReply {
fn from(r: hive_sh4re::Response) -> Self {
match r {
hive_sh4re::AgentResponse::Ok => Self::Ok,
hive_sh4re::AgentResponse::Err { message } => Self::Err(message),
hive_sh4re::AgentResponse::Messages { messages } => Self::Messages(messages),
hive_sh4re::AgentResponse::Status { unread } => Self::Status(unread),
hive_sh4re::AgentResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::AgentResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::AgentResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::AgentResponse::PendingRemindersCount { count } => {
hive_sh4re::Response::Ok => Self::Ok,
hive_sh4re::Response::Err { message } => Self::Err(message),
hive_sh4re::Response::Messages { messages } => Self::Messages(messages),
hive_sh4re::Response::Status { unread } => Self::Status(unread),
hive_sh4re::Response::Recent { rows } => Self::Recent(rows),
hive_sh4re::Response::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::Response::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::Response::PendingRemindersCount { count } => {
Self::PendingRemindersCount(count)
}
hive_sh4re::AgentResponse::ReminderRollup(stats) => Self::ReminderRollup(stats),
hive_sh4re::AgentResponse::AgentMeta {
name,
role,
running,
hyperhive_rev,
status_text,
status_set_at,
hive_name,
swarm_name,
} => Self::AgentMeta {
name,
role,
running,
hyperhive_rev,
status_text,
status_set_at,
hive_name,
swarm_name,
},
}
}
}
impl From<hive_sh4re::ManagerResponse> for SocketReply {
fn from(r: hive_sh4re::ManagerResponse) -> Self {
match r {
hive_sh4re::ManagerResponse::Ok => Self::Ok,
hive_sh4re::ManagerResponse::Err { message } => Self::Err(message),
hive_sh4re::ManagerResponse::Messages { messages } => Self::Messages(messages),
hive_sh4re::ManagerResponse::Status { unread } => Self::Status(unread),
hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content),
hive_sh4re::ManagerResponse::Schedules { schedules } => Self::Schedules(schedules),
hive_sh4re::ManagerResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::ManagerResponse::PendingRemindersCount { count } => {
Self::PendingRemindersCount(count)
}
hive_sh4re::ManagerResponse::ReminderRollup(stats) => Self::ReminderRollup(stats),
hive_sh4re::ManagerResponse::AgentMeta {
hive_sh4re::Response::ReminderRollup(stats) => Self::ReminderRollup(stats),
hive_sh4re::Response::Logs { content } => Self::Logs(content),
hive_sh4re::Response::Schedules { schedules } => Self::Schedules(schedules),
hive_sh4re::Response::AgentMeta {
name,
role,
running,
@ -647,7 +610,7 @@ impl AgentServer {
)]
async fn get_loose_ends(&self) -> String {
run_tool_envelope("get_loose_ends", String::new(), async move {
let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds).await;
let (resp, retries) = self.dispatch(hive_sh4re::AgentRequest::GetLooseEnds { agent: None }).await;
annotate_retries(format_loose_ends(resp), retries)
})
.await

167
hive-ag3nt/src/web_ui.rs
View file

@ -63,17 +63,7 @@ struct AppState {
gui_vnc_port: Option<u16>,
}
impl AppState {
fn flavor(&self) -> Flavor {
self.files.flavor
}
}
/// Which wire protocol the per-agent UI's `/send` handler should speak.
/// Sub-agent → `AgentRequest::OperatorMsg`; manager →
/// `ManagerRequest::OperatorMsg`. Reuses the MCP-side enum so a
/// single value drives both the send protocol and (in
/// `post_compact`) the allowed-tools surface claude sees.
/// Re-export so callers in `turn.rs` can name the type via `web_ui::Flavor`.
pub type Flavor = mcp::Flavor;
/// Bind the per-container web listener and serve the SPA.
@ -379,7 +369,7 @@ async fn api_stats(
// filters its counts to the same time range as the chart data.
let window_secs = window.span_secs();
let window_secs_u = u64::try_from(window_secs).unwrap_or(0);
snapshot.reminder_stats = fetch_reminder_stats(&state.socket, state.flavor(), window_secs_u).await;
snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await;
axum::Json(snapshot)
}
@ -505,39 +495,18 @@ struct SessionView {
/// the `mcp__hyperhive__get_loose_ends` tool sees from inside the
/// container.
async fn api_loose_ends(State(state): State<AppState>) -> Response {
let loose_ends: Vec<hive_sh4re::LooseEnd> = match state.flavor() {
Flavor::Agent => {
match client::request::<_, hive_sh4re::AgentResponse>(
&state.socket,
&hive_sh4re::AgentRequest::GetLooseEnds,
)
.await
{
Ok(hive_sh4re::AgentResponse::LooseEnds { loose_ends }) => loose_ends,
Ok(hive_sh4re::AgentResponse::Err { message }) => {
return error_response(&format!("get_loose_ends: {message}"));
}
Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
Err(e) => return error_response(&format!("transport: {e:#}")),
}
}
Flavor::Manager => {
match client::request::<_, hive_sh4re::ManagerResponse>(
&state.socket,
// Manager's own loose ends — the web page is the
// manager's page, not a hive-wide console.
&hive_sh4re::ManagerRequest::GetLooseEnds { agent: None },
)
.await
{
Ok(hive_sh4re::ManagerResponse::LooseEnds { loose_ends }) => loose_ends,
Ok(hive_sh4re::ManagerResponse::Err { message }) => {
return error_response(&format!("get_loose_ends: {message}"));
}
Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
Err(e) => return error_response(&format!("transport: {e:#}")),
}
let loose_ends: Vec<hive_sh4re::LooseEnd> = match client::request::<_, hive_sh4re::Response>(
&state.socket,
&hive_sh4re::Request::GetLooseEnds { agent: None },
)
.await
{
Ok(hive_sh4re::Response::LooseEnds { loose_ends }) => loose_ends,
Ok(hive_sh4re::Response::Err { message }) => {
return error_response(&format!("get_loose_ends: {message}"));
}
Ok(other) => return error_response(&format!("unexpected response: {other:?}")),
Err(e) => return error_response(&format!("transport: {e:#}")),
};
axum::Json(serde_json::json!({ "loose_ends": loose_ends })).into_response()
}
@ -567,7 +536,7 @@ async fn api_state(State(state): State<AppState>) -> axum::Json<StateSnapshot> {
.ok()
.and_then(|s| s.parse::<u16>().ok())
.unwrap_or(7000);
let inbox = recent_inbox(&state.socket, state.flavor()).await;
let inbox = recent_inbox(&state.socket).await;
let (turn_state, turn_state_since) = state.bus.state_snapshot();
let model = state.bus.model();
let context_window_tokens = state
@ -676,67 +645,34 @@ struct ExtraLink {
/// Best-effort: pull the last 30 messages addressed to us via the
/// per-agent / manager socket. Empty list on any transport / decode
/// failure — the inbox section is decorative, not authoritative.
async fn recent_inbox(socket: &std::path::Path, flavor: Flavor) -> Vec<hive_sh4re::InboxRow> {
async fn recent_inbox(socket: &std::path::Path) -> Vec<hive_sh4re::InboxRow> {
const LIMIT: u64 = 30;
match flavor {
Flavor::Agent => {
match client::request::<_, hive_sh4re::AgentResponse>(
socket,
&hive_sh4re::AgentRequest::Recent { limit: LIMIT },
)
.await
{
Ok(hive_sh4re::AgentResponse::Recent { rows }) => rows,
_ => Vec::new(),
}
}
Flavor::Manager => {
match client::request::<_, hive_sh4re::ManagerResponse>(
socket,
&hive_sh4re::ManagerRequest::Recent { limit: LIMIT },
)
.await
{
Ok(hive_sh4re::ManagerResponse::Recent { rows }) => rows,
_ => Vec::new(),
}
}
match client::request::<_, hive_sh4re::Response>(
socket,
&hive_sh4re::Request::Recent { limit: LIMIT },
)
.await
{
Ok(hive_sh4re::Response::Recent { rows }) => rows,
_ => Vec::new(),
}
}
/// Fetch reminder activity stats from the broker via the per-agent /
/// manager socket. Returns None on any transport / decode failure — the
/// stats are decorative, not authoritative.
async fn fetch_reminder_stats(socket: &std::path::Path, flavor: Flavor, window_secs: u64) -> Option<hive_sh4re::ReminderStats> {
match flavor {
Flavor::Agent => {
match client::request::<_, hive_sh4re::AgentResponse>(
socket,
&hive_sh4re::AgentRequest::ReminderRollup {
since_secs: window_secs,
},
)
.await
{
Ok(hive_sh4re::AgentResponse::ReminderRollup(stats)) => Some(stats),
_ => None,
}
}
Flavor::Manager => {
match client::request::<_, hive_sh4re::ManagerResponse>(
socket,
&hive_sh4re::ManagerRequest::ReminderRollup {
since_secs: window_secs,
// Manager's own stats page — its own reminders.
agent: None,
},
)
.await
{
Ok(hive_sh4re::ManagerResponse::ReminderRollup(stats)) => Some(stats),
_ => None,
}
}
async fn fetch_reminder_stats(socket: &std::path::Path, window_secs: u64) -> Option<hive_sh4re::ReminderStats> {
match client::request::<_, hive_sh4re::Response>(
socket,
&hive_sh4re::Request::ReminderRollup {
since_secs: window_secs,
agent: None,
},
)
.await
{
Ok(hive_sh4re::Response::ReminderRollup(stats)) => Some(stats),
_ => None,
}
}
@ -754,29 +690,16 @@ async fn post_send(State(state): State<AppState>, Form(form): Form<SendForm>) ->
if body.is_empty() {
return error_response("send: `body` required");
}
let result = match state.flavor() {
Flavor::Agent => match client::request::<_, hive_sh4re::AgentResponse>(
&state.socket,
&hive_sh4re::AgentRequest::OperatorMsg { body },
)
.await
{
Ok(hive_sh4re::AgentResponse::Ok) => Ok(()),
Ok(hive_sh4re::AgentResponse::Err { message }) => Err(message),
Ok(other) => Err(format!("unexpected response: {other:?}")),
Err(e) => Err(format!("transport: {e:#}")),
},
Flavor::Manager => match client::request::<_, hive_sh4re::ManagerResponse>(
&state.socket,
&hive_sh4re::ManagerRequest::OperatorMsg { body },
)
.await
{
Ok(hive_sh4re::ManagerResponse::Ok) => Ok(()),
Ok(hive_sh4re::ManagerResponse::Err { message }) => Err(message),
Ok(other) => Err(format!("unexpected response: {other:?}")),
Err(e) => Err(format!("transport: {e:#}")),
},
let result = match client::request::<_, hive_sh4re::Response>(
&state.socket,
&hive_sh4re::Request::OperatorMsg { body },
)
.await
{
Ok(hive_sh4re::Response::Ok) => Ok(()),
Ok(hive_sh4re::Response::Err { message }) => Err(message),
Ok(other) => Err(format!("unexpected response: {other:?}")),
Err(e) => Err(format!("transport: {e:#}")),
};
match result {
// 200 instead of 303 → the client doesn't refetch /api/state.