vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

agents: remove send(operator) and the operator inbox; alert on turn failures

Browse source 
Agents reach the operator over matrix now (#5115). A `send` to
`operator` is refused, by the broker and by the MCP allow-list check,
with an error that names the matrix MCP tools. The dashboard's Y3R C4LL
tab goes with it: the operator inbox, `GET /api/operator-inbox`,
`Broker::unread_for_recipient`, and the browser notifications that only
ever fired on operator-bound messages.

The harness no longer posts turn failures and plugin-install failures
to the operator inbox. Both are already logged at warn; a new Grafana
rule, `hh-agent-turn-failed`, fires per agent on the harness's
`turn failed` line.

Dashboard-to-agent messaging (op-send, the agent web UI's message box,
mark-all-read for agent inboxes) is unchanged.

Decisions per mara on #5115: q1 a, q2 a plus a grafana alert, q3 keep
dashboard-to-agent messaging and leave swarm-level code alone.
This commit is contained in:
atlas 2026-10-12 00:44:11 +02:00 • committed by mara
commit 7e6a4c7727
37 changed files with 246 additions and 848 deletions
Download patch file Download diff file Expand all files Collapse all files

80
hive-c0re/src/dashboard/misc_api.rs
View file

@ -1,6 +1,6 @@
//! Remaining single-endpoint dashboard handlers: the operator inbox
//! (`Y3R C4LL`) + mark-all-read, operator compose (`op-send`),
//! spawn-request, hive-wide turn stats, and container resources.
//! Remaining single-endpoint dashboard handlers: mark-all-read, operator
//! compose (`op-send`), spawn-request, hive-wide turn stats, and container
//! resources.
use axum::{
extract::{Form, Path as AxumPath, State},
@ -10,80 +10,10 @@ use axum::{
use serde::{Deserialize, Serialize};
use utoipa::{IntoParams, ToSchema};
use super::{AppState, Ident, error_response, scan_validated_paths};
use super::{AppState, Ident, error_response};
use crate::container_stats::ContainerResource;
use crate::hive_stats::HiveStats;
#[derive(Serialize, ToSchema)]
pub(super) struct OperatorInboxItem {
id: i64,
from: String,
body: String,
at: chrono::DateTime<chrono::Utc>,
in_reply_to: Option<i64>,
file_refs: Vec<String>,
}
#[derive(Serialize, ToSchema)]
pub(super) struct OperatorInboxBody {
messages: Vec<OperatorInboxItem>,
}
/// Unread operator-directed messages for the dashboard's Y3R C4LL inbox.
///
/// Returns messages addressed to `"operator"` that haven't been
/// acked yet (the operator clears them via the existing
/// `POST /api/agent/operator/mark-all-read`). Newest-first; path-shaped
/// tokens are validated so the client renders file links like the
/// terminal does.
#[utoipa::path(
get,
path = "/api/operator-inbox",
responses(
(status = 200, description = "unread operator-directed messages", body = OperatorInboxBody),
(status = 500, description = "broker read failed"),
),
tag = "misc_api"
)]
pub(super) async fn api_operator_inbox(State(state): State<AppState>) -> Response {
const INBOX_LIMIT: u64 = 100;
match state
.coord
.broker
.unread_for_recipient("operator", INBOX_LIMIT)
{
Ok(messages) => {
let messages: Vec<OperatorInboxItem> = messages
.into_iter()
.filter_map(|m| {
let crate::broker::MessageEvent::Sent {
id,
from,
body,
at,
in_reply_to,
..
} = m
else {
return None;
};
let file_refs = scan_validated_paths(&body);
Some(OperatorInboxItem {
id,
from,
at: hive_sh4re::wire_time::from_secs(at),
body,
in_reply_to,
file_refs,
})
})
.collect();
axum::Json(OperatorInboxBody { messages }).into_response()
}
Err(e) => error_response(&format!("operator-inbox failed: {e:#}")),
}
}
#[derive(Deserialize, IntoParams)]
pub(super) struct StatsHiveQuery {
/// Stats window; defaults to `24h`.
@ -227,7 +157,7 @@ pub(super) async fn post_op_send(
// 200 instead of 303 → the client doesn't refetch /api/state. The
// broker `send` already emitted a `MessageEvent` which the
// dashboard channel forwarder mirrors as `DashboardEvent::Sent`,
// and the page's terminal + inbox derive from that stream — so the
// and the page's terminal derives from that stream — so the
// operator's send shows up the same way an agent's send does, with
// no full-state refresh in between.
(axum::http::StatusCode::OK, "ok").into_response()

56
hive-c0re/src/dashboard/mod.rs
View file

@ -36,7 +36,7 @@ use crate::lifecycle;
(name = "build_logs", description = "build log headers, full rows, and raw text downloads"),
(name = "lifecycle_ops", description = "agent container lifecycle: rebuild/restart/start/stop/pause/limits"),
(name = "meta_inputs", description = "bulk flake-input update for the meta flake"),
(name = "misc_api", description = "operator inbox, compose, spawn-request, hive stats"),
(name = "misc_api", description = "mark-all-read, compose, spawn-request, hive stats"),
(name = "permissions", description = "tool-group + capability assignment for agents"),
(name = "schedules", description = "read-only scheduled-prompt list + rebuild-queue cancel"),
(name = "state_files", description = "proxied reads of allow-listed per-agent state files"),
@ -89,7 +89,7 @@ struct AppState {
handler; splitting that exhaustive list across helpers would \
obscure the route map for no readability gain"
)]
pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
fn app(coord: Arc<Coordinator>) -> axum::Router {
// API-only: the gateway static-serves the dashboard dist and proxies
// non-static requests here (see `nix/host-modules/hive-gateway/vhosts.nix`).
// Unmatched paths 404.
@ -115,7 +115,6 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
.routes(routes!(health::get_health_ready))
.routes(routes!(state_snapshot::api_state))
.routes(routes!(state_files::get_state_file))
.routes(routes!(misc_api::api_operator_inbox))
.routes(routes!(misc_api::api_stats_hive))
.routes(routes!(misc_api::api_container_resources))
.routes(routes!(misc_api::post_mark_all_read))
@ -156,18 +155,21 @@ pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
// the nix store (see the module doc comment above `ApiDoc`). `api`
// is `Clone`; each request gets its own owned copy for `Json` to
// serialize.
let app = router
router
.route(
"/api/openapi.json",
get(move || async move { Json(api.clone()) }),
)
.with_state(AppState { coord });
.with_state(AppState { coord })
}
pub async fn serve(port: u16, coord: Arc<Coordinator>) -> Result<()> {
// Binds loopback-only; external access via gateway.
// Rationale: docs/networking/gateway.md::Firewall posture.
let addr = SocketAddr::from(([127, 0, 0, 1], port));
let listener = bind_with_retry(addr).await?;
tracing::info!(%addr, "dashboard listening");
axum::serve(listener, app).await?;
axum::serve(listener, app(coord)).await?;
Ok(())
}
@ -287,6 +289,47 @@ fn error_response(message: &str) -> Response {
mod tests {
use super::*;
/// Status code of a bare HTTP/1.1 GET against the dashboard's own route
/// table, served on an ephemeral loopback port.
async fn status_of(path: &str) -> u16 {
use tokio::io::{AsyncReadExt, AsyncWriteExt};
let dir = tempfile::tempdir().expect("tempdir");
let coord = Coordinator::open(
&dir.path().join("broker.sqlite"),
crate::coordinator::HiveEnv::default(),
crate::hive_stats::PriceTable::new(),
1,
)
.expect("open coordinator");
let listener = tokio::net::TcpListener::bind("127.0.0.1:0")
.await
.expect("bind");
let addr = listener.local_addr().expect("local addr");
let server = tokio::spawn(axum::serve(listener, app(Arc::new(coord))).into_future());
let mut stream = tokio::net::TcpStream::connect(addr).await.expect("connect");
stream
.write_all(
format!("GET {path} HTTP/1.1\r\nHost: t\r\nConnection: close\r\n\r\n").as_bytes(),
)
.await
.expect("write");
let mut reply = String::new();
stream.read_to_string(&mut reply).await.expect("read");
server.abort();
reply
.split_whitespace()
.nth(1)
.and_then(|code| code.parse().ok())
.unwrap_or_else(|| panic!("no status line in {reply:?}"))
}
#[tokio::test]
async fn operator_inbox_route_is_gone() {
assert_eq!(status_of("/health/live").await, 200);
assert_eq!(status_of("/api/operator-inbox").await, 404);
}
#[test]
fn problem_details_carry_rfc9457_status_and_detail() {
// Contract the frontend depends on: the problem_details crate
@ -329,7 +372,6 @@ mod router_build_probe {
.routes(routes!(health::get_health_ready))
.routes(routes!(state_snapshot::api_state))
.routes(routes!(state_files::get_state_file))
.routes(routes!(misc_api::api_operator_inbox))
.routes(routes!(misc_api::api_stats_hive))
.routes(routes!(misc_api::api_container_resources))
.routes(routes!(misc_api::post_mark_all_read))

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

@ -491,7 +491,7 @@ fn handle_status(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_sock:
}
/// `OperatorMsg` — deliver an operator-authored message into `agent`'s
/// inbox (from the `operator` recipient).
/// inbox (sender `operator`).
fn handle_operator_msg(
coord: &Arc<Coordinator>,
agent: &str,
@ -603,12 +603,11 @@ pub(crate) fn handle_send(
}
};
}
// Validate that the recipient is a known local agent or the
// special "operator" recipient. Without this check a typo in `to`
// silently queues a message nobody will ever read.
// Validate that the recipient is a known local agent. Without this
// check a typo in `to` silently queues a message nobody will ever read.
//
// Cross-hive messaging (`name@hive` qualified names) is not routed
// through the broker — use the Matrix MCP tools for that instead.
// Cross-hive messaging (`name@hive` qualified names) and the operator
// are not reached through the broker — the Matrix MCP tools are.
if to.contains('@') {
return Response::Err {
message: format!(
@ -617,20 +616,23 @@ pub(crate) fn handle_send(
),
};
}
if to != hive_sh4re::manager::OPERATOR_RECIPIENT {
// A name that doesn't parse as an Ident can't be a local agent, so
// it collapses into the same "unknown recipient" error as a valid
// name with no state dir.
let exists = hive_types::Ident::parse(to)
.is_ok_and(|id| crate::paths::agent_state_dir(&id).exists());
if !exists {
return Response::Err {
message: format!(
"send failed: unknown recipient `{to}` \
(no agent with that name exists on this hive)"
),
};
}
if to == hive_sh4re::manager::OPERATOR_RECIPIENT {
return Response::Err {
message: hive_sh4re::manager::OPERATOR_SEND_REFUSAL.to_owned(),
};
}
// A name that doesn't parse as an Ident can't be a local agent, so
// it collapses into the same "unknown recipient" error as a valid
// name with no state dir.
let exists =
hive_types::Ident::parse(to).is_ok_and(|id| crate::paths::agent_state_dir(&id).exists());
if !exists {
return Response::Err {
message: format!(
"send failed: unknown recipient `{to}` \
(no agent with that name exists on this hive)"
),
};
}
match coord.broker.send(&Message {
from: hive_sh4re::manager::trusted_sender(agent),
@ -664,6 +666,22 @@ mod tests {
(dir, Arc::new(coord))
}
#[test]
fn send_to_operator_is_refused_with_the_matrix_pointer() {
let (_dir, coord) = coordinator();
let resp = handle_send(
&coord,
"iris",
hive_sh4re::manager::OPERATOR_RECIPIENT,
"hi",
None,
);
assert!(
matches!(&resp, Response::Err { message } if message.contains("mcp__matrix__")),
"{resp:?}"
);
}
#[tokio::test]
async fn over_long_line_is_rejected_after_reading_only_the_bound() {
let total = 2 * REQUEST_LINE_MAX_BYTES as u64;

38
hive-c0re/src/stores/broker.rs
View file

@ -301,44 +301,6 @@ impl Broker {
.map_err(Into::into)
}
/// Unacknowledged messages addressed to `recipient`, newest-first.
/// Backs the dashboard's operator inbox: the operator never
/// `recv`s over an agent socket, so messages to `"operator"` sit in
/// the broker with `acked_at IS NULL` until the operator hits "mark
/// all read" (which calls [`Broker::mark_all_read`]). This read
/// mirrors that filter EXACTLY — `recipient = ?1 AND acked_at IS
/// NULL`, with no `delivered_at` condition — so everything listed
/// here is precisely what `mark_all_read` will clear (operator rows
/// may never get `delivered_at` set). Returned as
/// [`MessageEvent::Sent`] so the dashboard reuses its live renderer.
///
/// # Errors
///
/// Returns `Err` if the `SQLite` prepare or query fails.
pub fn unread_for_recipient(&self, recipient: &str, limit: u64) -> Result<Vec<MessageEvent>> {
let conn = self.conn.lock().unwrap();
let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX);
let mut stmt = conn.prepare(
"SELECT id, sender, recipient, body, sent_at, in_reply_to
FROM messages
WHERE recipient = ?1 AND acked_at IS NULL
ORDER BY id DESC
LIMIT ?2",
)?;
let rows = stmt.query_map(params![recipient, limit_i], |row| {
Ok(MessageEvent::Sent {
id: row.get(0)?,
from: row.get(1)?,
to: row.get(2)?,
body: row.get(3)?,
at: row.get(4)?,
in_reply_to: row.get(5)?,
})
})?;
rows.collect::<rusqlite::Result<Vec<_>>>()
.map_err(Into::into)
}
/// Number of undelivered messages addressed to `recipient`. Non-mutating
/// — used by the harness to surface "N unread" in tool-result status
/// lines without popping the queue.