d94bc2188d
`topology.json` was a map of `name -> parent | null`, and that value fed the whole agent hierarchy: `<parent>` / `<children>` recipient sentinels, the reparenting API (CLI verb, wire verb, dashboard endpoints, DAG node), the dashboard tree, the rebuild depth sort, and an unconditional bind-mount grant giving every agent RW on its direct children's state. Per the operator's ruling the field goes, and with it all of the above. The file survives as what remains once the value is gone: the roster of agent names, which is the set `ManageRootAgent` grants mounts over. It is now a JSON array; `read` still accepts the old map shape and keeps its keys, so a hive that upgrades across this does not blank its roster (and so no capability holder loses its mounts for the length of that window). Two sites kept their behaviour under a different recipient rather than losing it. Both addressed `<parent>`, which the broker already resolved to `operator` for a root agent, and every agent is now what that fallback called a root: - the harness's turn-failure / plugin-failure notification (`Surface::send_to_parent` -> `send_to_operator`), and - the send allow-list's always-permitted escape hatch, so an agent with a restrictive allow-list still has a way to say it is stuck. What is NOT preserved, deliberately: an agent with no capability no longer sees any other agent's dirs. `ManageRootAgent`'s own grant is unchanged -- still every agent in the roster, still state RW + config RO, still no `harness`. The dashboard's reparenting control (the M0V3 picker) is deleted with its CSS. The tree rendering that reads `ContainerView.parent` is left for the frontend owner -- it degrades to a flat list with the field gone.
795 lines
32 KiB
Rust
795 lines
32 KiB
Rust
//! Unix-socket request server, shared by the per-agent sockets and the
|
|
//! (pure-transport) manager socket. The socket file's existence on disk
|
|
//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means
|
|
//! you are `foo`; the manager socket simply serves as `ruth`. There is no
|
|
//! privilege flag — both transports run the same [`serve`] / [`dispatch`]
|
|
//! code, and authority derives uniformly from the caller's identity:
|
|
//! capabilities for the hive-wide queries and tool-group membership for the
|
|
//! orchestration verbs. An agent-targeting verb may name any agent, so `ruth`
|
|
//! reaches every agent exactly the way every other agent does, not via any
|
|
//! hardcoded name match.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_core_agent_sock::{Request, Response};
|
|
use hive_sh4re::inbox::Message;
|
|
use hive_sh4re::manager::MANAGER_AGENT;
|
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
use tokio::task::JoinHandle;
|
|
|
|
use crate::coordinator::Coordinator;
|
|
|
|
mod config_approvals;
|
|
mod schedules;
|
|
|
|
pub(crate) use config_approvals::submit_merge_config_pr;
|
|
pub(crate) use schedules::filter_ghost_schedule_targets;
|
|
pub use schedules::schedule_to_wire_public;
|
|
|
|
use schedules::{
|
|
EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now,
|
|
handle_list_schedules, handle_request_schedule_prompt,
|
|
};
|
|
|
|
pub struct AgentSocket {
|
|
pub path: PathBuf,
|
|
pub handle: JoinHandle<()>,
|
|
}
|
|
|
|
pub fn start(agent: &str, socket_path: &Path, coord: Arc<Coordinator>) -> Result<AgentSocket> {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
let agent = agent.to_owned();
|
|
if let Some(parent) = socket_path.parent() {
|
|
std::fs::create_dir_all(parent)
|
|
.with_context(|| format!("create agent socket dir {}", parent.display()))?;
|
|
}
|
|
if socket_path.exists() {
|
|
std::fs::remove_file(socket_path).context("remove stale agent socket")?;
|
|
}
|
|
let listener = UnixListener::bind(socket_path)
|
|
.with_context(|| format!("bind agent socket {}", socket_path.display()))?;
|
|
// The socket is bind-mounted into exactly one container as
|
|
// `/run/hive/mcp.sock` (`lifecycle::set_nspawn_flags`); the
|
|
// in-container harness connects as the per-agent unix user,
|
|
// not root, so the default `tokio::net::UnixListener::bind`
|
|
// perms (0755) lock it out. 0666 lets the agent user connect;
|
|
// the bind source dir is per-agent on host so blast radius is
|
|
// unchanged.
|
|
std::fs::set_permissions(socket_path, std::fs::Permissions::from_mode(0o666))
|
|
.with_context(|| format!("chmod agent socket {}", socket_path.display()))?;
|
|
tracing::info!(%agent, socket = %socket_path.display(), "agent socket listening");
|
|
|
|
let path = socket_path.to_path_buf();
|
|
let handle = tokio::spawn(async move {
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((stream, _)) => {
|
|
let agent = agent.clone();
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = serve(stream, agent, coord).await {
|
|
tracing::warn!(error = ?e, "agent connection failed");
|
|
}
|
|
});
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "agent listener accept failed; exiting");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
Ok(AgentSocket { path, handle })
|
|
}
|
|
|
|
/// Bind + serve the manager socket. This is now **pure transport**: it grants
|
|
/// no authority of its own — it just serves requests as `agent = MANAGER_AGENT`
|
|
/// ("ruth"), and ruth's reach comes entirely from the capabilities /
|
|
/// tool-groups it holds, identical to connecting on a per-agent socket — ruth
|
|
/// uses the standard per-agent runtime dir + socket, with no dedicated helpers.
|
|
pub fn start_manager(coord: Arc<Coordinator>) -> Result<()> {
|
|
use std::os::unix::fs::PermissionsExt as _;
|
|
let dir = crate::paths::agent_runtime_dir(crate::lifecycle::MANAGER_NAME);
|
|
std::fs::create_dir_all(&dir)
|
|
.with_context(|| format!("create manager dir {}", dir.display()))?;
|
|
let socket = Coordinator::socket_path(crate::lifecycle::MANAGER_NAME);
|
|
if socket.exists() {
|
|
std::fs::remove_file(&socket).context("remove stale manager socket")?;
|
|
}
|
|
let listener = UnixListener::bind(&socket)
|
|
.with_context(|| format!("bind manager socket {}", socket.display()))?;
|
|
// 0666 so the in-container root user (non-root) can connect; the bind
|
|
// source dir is manager-only on host (see the per-agent socket above).
|
|
std::fs::set_permissions(&socket, std::fs::Permissions::from_mode(0o666))
|
|
.with_context(|| format!("chmod manager socket {}", socket.display()))?;
|
|
tracing::info!(socket = %socket.display(), "manager socket listening");
|
|
|
|
tokio::spawn(async move {
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((stream, _)) => {
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
// Pure transport: serve as `ruth`, no privilege grant.
|
|
if let Err(e) = serve(stream, MANAGER_AGENT.to_owned(), coord).await {
|
|
tracing::warn!(error = ?e, "manager connection failed");
|
|
}
|
|
});
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "manager listener accept failed");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
Ok(())
|
|
}
|
|
|
|
async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
|
|
let (read, mut write) = stream.into_split();
|
|
let mut reader = BufReader::new(read);
|
|
let mut line = String::new();
|
|
loop {
|
|
line.clear();
|
|
let n = reader.read_line(&mut line).await?;
|
|
if n == 0 {
|
|
return Ok(());
|
|
}
|
|
let resp = match serde_json::from_str::<Request>(line.trim()) {
|
|
Ok(req) => dispatch(&req, &agent, &coord).await,
|
|
Err(e) => Response::Err {
|
|
message: format!("parse error: {e}"),
|
|
},
|
|
};
|
|
let mut payload = serde_json::to_string(&resp)?;
|
|
payload.push('\n');
|
|
write.write_all(payload.as_bytes()).await?;
|
|
write.flush().await?;
|
|
}
|
|
}
|
|
|
|
/// Max long-poll window the caller can ask for; values above the
|
|
/// cap are clamped. 180s keeps us under typical TCP/proxy idle
|
|
/// limits while still letting agents park their turn until a
|
|
/// message arrives. Omitting `wait_seconds` (or passing `0`) means
|
|
/// "peek, don't wait" — claude can call recv whenever it wants a
|
|
/// cheap "is there anything pending?" check without blocking the
|
|
/// turn for 30 seconds. To actually park, the caller passes a
|
|
/// positive `wait_seconds`.
|
|
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_mins(3);
|
|
|
|
/// Server-side hard cap on `Recv.max` — canonical value lives in
|
|
/// `hive_sh4re::inbox::RECV_BATCH_MAX` so the harness's wake-prompt hint and
|
|
/// this enforcement site can't drift apart.
|
|
pub(crate) const RECV_BATCH_MAX: u32 = hive_sh4re::inbox::RECV_BATCH_MAX;
|
|
|
|
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,
|
|
}
|
|
}
|
|
|
|
/// 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 the remaining variants —
|
|
/// `GetLooseEnds` and the orchestration verbs (schedules), which need
|
|
/// per-verb tool-group gating — or for host-admin / unknown requests
|
|
/// invalid on either socket.
|
|
///
|
|
/// The unified `dispatch` calls this first; the remaining arms (which gate
|
|
/// on topology / capabilities / tool-groups) are handled there.
|
|
pub(crate) async fn dispatch_shared(
|
|
req: &hive_core_agent_sock::Request,
|
|
agent: &str,
|
|
coord: &Arc<Coordinator>,
|
|
) -> Option<hive_core_agent_sock::Response> {
|
|
Some(match req {
|
|
hive_core_agent_sock::Request::Send {
|
|
to,
|
|
body,
|
|
in_reply_to,
|
|
} => handle_send(coord, agent, to, body, *in_reply_to),
|
|
hive_core_agent_sock::Request::Recv { wait_seconds, max } => {
|
|
handle_recv(coord, agent, *wait_seconds, *max).await
|
|
}
|
|
hive_core_agent_sock::Request::Status => handle_status(coord, agent),
|
|
hive_core_agent_sock::Request::OperatorMsg { body } => {
|
|
handle_operator_msg(coord, agent, body)
|
|
}
|
|
hive_core_agent_sock::Request::Wake { from, body } => handle_wake(coord, agent, from, body),
|
|
hive_core_agent_sock::Request::Recent { limit } => handle_recent(coord, agent, *limit),
|
|
hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text),
|
|
hive_core_agent_sock::Request::GetAgentMeta { name } => {
|
|
handle_get_agent_meta(coord, agent, name.as_ref()).await
|
|
}
|
|
hive_core_agent_sock::Request::CancelLooseEnd { kind, id } => {
|
|
handle_cancel_loose_end(coord, agent, *kind, *id).map_or_else(
|
|
|message| hive_core_agent_sock::Response::Err { message },
|
|
|()| hive_core_agent_sock::Response::Ok,
|
|
)
|
|
}
|
|
hive_core_agent_sock::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await,
|
|
hive_core_agent_sock::Request::AckTurn => handle_ack_turn(coord, agent),
|
|
hive_core_agent_sock::Request::AckUntil { up_to } => handle_ack_until(coord, agent, *up_to),
|
|
hive_core_agent_sock::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
|
|
hive_core_agent_sock::Request::GracefulStopComplete => {
|
|
// Harness drained + is exiting: clear the fence so the
|
|
// `GracefulStop` orchestration (which polls this flag) proceeds
|
|
// to stop the container without waiting out its timeout.
|
|
coord.clear_graceful_stop(agent);
|
|
hive_core_agent_sock::Response::Ok
|
|
}
|
|
hive_core_agent_sock::Request::PauseAcknowledged => {
|
|
// Harness saw its own pause marker flip between turns: clear
|
|
// the fence so the pause DAG's drain node (which polls this
|
|
// flag) resolves without waiting out its timeout.
|
|
coord.clear_pause_pending(agent);
|
|
hive_core_agent_sock::Response::Ok
|
|
}
|
|
// Not a shared variant.
|
|
_ => return None,
|
|
})
|
|
}
|
|
|
|
/// `Recv` — long-poll the broker for up to `max` messages (capped at
|
|
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
|
|
async fn handle_recv(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
wait_seconds: Option<u64>,
|
|
max: Option<u32>,
|
|
) -> hive_core_agent_sock::Response {
|
|
// Graceful-stop fence: while a graceful stop is pending for this agent,
|
|
// return `GracefulStop` instead of polling the broker. The harness runs
|
|
// one stop-checkpoint turn then exits; new sends keep queueing in the
|
|
// broker for the agent's next start. Checked before the (blocking) poll
|
|
// so a flag set between polls is seen on the next Recv — the orchestration
|
|
// also fires a transient wake to break an in-flight long-poll.
|
|
if coord.is_graceful_stop_pending(agent) {
|
|
return hive_core_agent_sock::Response::GracefulStop;
|
|
}
|
|
let cap = max.unwrap_or(1).min(RECV_BATCH_MAX) as usize;
|
|
match coord
|
|
.broker
|
|
.recv_blocking_batch(agent, recv_timeout(wait_seconds), cap)
|
|
.await
|
|
{
|
|
Ok(deliveries) => {
|
|
// `recv_batch` stamps `delivered_at` on every popped row, so a
|
|
// `count_pending` here excludes the just-popped batch and reports
|
|
// exactly how many still-pending messages remain to drain. A count
|
|
// error is non-fatal — fall back to 0 rather than fail the recv.
|
|
let remaining = coord.broker.count_pending(agent).unwrap_or(0);
|
|
hive_core_agent_sock::Response::Messages {
|
|
messages: deliveries
|
|
.into_iter()
|
|
.map(|d| hive_sh4re::inbox::DeliveredMessage {
|
|
from: d.message.from.to_string(),
|
|
body: d.message.body,
|
|
id: d.id,
|
|
redelivered: d.redelivered,
|
|
in_reply_to: d.message.in_reply_to,
|
|
})
|
|
.collect(),
|
|
remaining,
|
|
}
|
|
}
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Wake` — inject a wake into `agent`'s own inbox. Persisted through
|
|
/// the sqlite broker like any other message so the agent can ack it
|
|
/// via `AckUntil` and it appears in message history for post-mortem.
|
|
fn handle_wake(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
from: &str,
|
|
body: &str,
|
|
) -> hive_core_agent_sock::Response {
|
|
match coord.broker.send(&Message {
|
|
from: hive_sh4re::manager::trusted_sender(from),
|
|
to: agent.to_owned(),
|
|
body: body.to_owned(),
|
|
in_reply_to: None,
|
|
}) {
|
|
Ok(()) => hive_core_agent_sock::Response::Ok,
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `SetStatus` — validate the status text, then trigger a dashboard
|
|
/// rescan. The harness has already written the status file to its own
|
|
/// `state/` dir (it runs as the agent user), so this only refreshes the
|
|
/// dashboard's view.
|
|
fn handle_set_status(coord: &Arc<Coordinator>, text: &str) -> hive_core_agent_sock::Response {
|
|
if let Err(message) = crate::limits::check_status_text(text) {
|
|
return hive_core_agent_sock::Response::Err { message };
|
|
}
|
|
let coord2 = Arc::clone(coord);
|
|
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
|
|
hive_core_agent_sock::Response::Ok
|
|
}
|
|
|
|
/// Validate an agent-supplied repo name: a single safe slug segment, no
|
|
/// path traversal. Forgejo validates server-side too, but rejecting early
|
|
/// gives a clear message and avoids building odd API paths.
|
|
fn valid_repo_name(name: &str) -> bool {
|
|
!name.is_empty()
|
|
&& name.len() <= 100
|
|
&& !name.starts_with(['-', '.'])
|
|
&& name
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
|
|
}
|
|
|
|
/// `CreateRepo` — create a repo for `agent` *through hive-c0re* in the
|
|
/// c0re-owned `agents` org with operator-team branch protection.
|
|
/// The sanctioned create path now that agents can't create repos directly.
|
|
async fn handle_create_repo(agent: &str, repo: &str) -> hive_core_agent_sock::Response {
|
|
if !valid_repo_name(repo) {
|
|
return hive_core_agent_sock::Response::Err {
|
|
message: format!(
|
|
"invalid repo name {repo:?} — single segment of letters, digits, '-', '_', '.' \
|
|
(no leading '-'/'.', max 100 chars)"
|
|
),
|
|
};
|
|
}
|
|
let Some(core_token) = crate::forge::core_token() else {
|
|
return hive_core_agent_sock::Response::Err {
|
|
message: "forge unavailable (no core token) — cannot create repo".to_owned(),
|
|
};
|
|
};
|
|
match crate::forge::create_agent_repo(agent, repo, &core_token).await {
|
|
Ok(full_name) => hive_core_agent_sock::Response::RepoCreated {
|
|
clone_url: format!("{}/{full_name}.git", crate::forge::forge_http_base()),
|
|
full_name,
|
|
},
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("create repo {repo:?} failed: {e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `GetAgentMeta` — identity + live status for `name` (defaults to the
|
|
/// caller). Reads the live container-view status and the hive/swarm
|
|
/// display names.
|
|
async fn handle_get_agent_meta(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
name: Option<&hive_types::Ident>,
|
|
) -> hive_core_agent_sock::Response {
|
|
// `name` arrives pre-validated by serde (wire field is `Ident`). The
|
|
// `None` default (target == caller) still needs a parse since `agent`
|
|
// is a plain `&str` here — but it's the caller's own authenticated
|
|
// name, already valid in practice.
|
|
let target_id = match name {
|
|
Some(id) => id.clone(),
|
|
None => match hive_types::Ident::parse(agent) {
|
|
Ok(id) => id,
|
|
Err(reason) => {
|
|
return hive_core_agent_sock::Response::Err {
|
|
message: format!("get_agent_meta: invalid agent name {agent:?}: {reason}"),
|
|
};
|
|
}
|
|
},
|
|
};
|
|
let (status_text, status_set_at, running) =
|
|
crate::container_view::read_agent_status_live(&target_id).await;
|
|
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
|
|
// Matrix identities are public handles (`name` / `user_id`
|
|
// `@user:server` / `homeserver`) — the access token lives separately
|
|
// in the agent's `matrix-token` and is never part of this response.
|
|
// Peer visibility is intentional: it lets an agent verify/contact
|
|
// another on a public matrix instance. The `Ident::parse` gate
|
|
// above is what closes the real vector here (path traversal via `../`
|
|
// in an agent-supplied name).
|
|
let matrix_accounts = read_agent_matrix_identities(&target_id);
|
|
hive_core_agent_sock::Response::AgentMeta {
|
|
name: target_id.into_string(),
|
|
running,
|
|
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
|
status_text,
|
|
status_set_at,
|
|
hive_name,
|
|
swarm_name,
|
|
matrix_accounts,
|
|
}
|
|
}
|
|
|
|
/// Read the target agent's matrix identities from the daemon's
|
|
/// `matrix-accounts.json` snapshot (under the agent's state dir).
|
|
/// Best-effort: an absent / unparseable snapshot (no matrix provisioning,
|
|
/// or the daemon not up yet) yields an empty list. The `MatrixIdentity`
|
|
/// serde shape matches the snapshot entries; the snapshot's `live` field is
|
|
/// ignored (only live accounts are written).
|
|
fn read_agent_matrix_identities(
|
|
agent: &hive_types::Ident,
|
|
) -> Vec<hive_sh4re::container::MatrixIdentity> {
|
|
let path = Coordinator::agent_notes_dir(agent).join("matrix-accounts.json");
|
|
std::fs::read_to_string(&path)
|
|
.ok()
|
|
.and_then(|s| serde_json::from_str::<Vec<hive_sh4re::container::MatrixIdentity>>(&s).ok())
|
|
.unwrap_or_default()
|
|
}
|
|
|
|
/// `Status` — count of pending (unread) inbox messages for `agent`.
|
|
fn handle_status(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_sock::Response {
|
|
match coord.broker.count_pending(agent) {
|
|
Ok(unread) => hive_core_agent_sock::Response::Status { unread },
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `OperatorMsg` — deliver an operator-authored message into `agent`'s
|
|
/// inbox (from the `operator` recipient).
|
|
fn handle_operator_msg(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
body: &str,
|
|
) -> hive_core_agent_sock::Response {
|
|
match coord.broker.send(&Message {
|
|
from: hive_sh4re::manager::trusted_sender(hive_sh4re::manager::OPERATOR_RECIPIENT),
|
|
to: agent.to_owned(),
|
|
body: body.to_owned(),
|
|
in_reply_to: None,
|
|
}) {
|
|
Ok(()) => hive_core_agent_sock::Response::Ok,
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Recent` — the last `limit` inbox rows for `agent` (read-only,
|
|
/// doesn't consume).
|
|
fn handle_recent(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
limit: u64,
|
|
) -> hive_core_agent_sock::Response {
|
|
match coord.broker.recent_for(agent, limit) {
|
|
Ok(rows) => hive_core_agent_sock::Response::Recent { rows },
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `AckTurn` — mark `agent`'s in-flight delivered messages acked so
|
|
/// they don't redeliver on the next turn.
|
|
fn handle_ack_turn(coord: &Arc<Coordinator>, agent: &str) -> hive_core_agent_sock::Response {
|
|
match coord.broker.ack_turn(agent) {
|
|
Ok(_n) => hive_core_agent_sock::Response::Ok,
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `AckUntil` — bulk-ack every message addressed to `agent` with row
|
|
/// id `<= up_to` (the agent-side backlog-triage escape hatch).
|
|
fn handle_ack_until(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
up_to: i64,
|
|
) -> hive_core_agent_sock::Response {
|
|
match coord.broker.ack_until(agent, up_to) {
|
|
Ok(count) => hive_core_agent_sock::Response::Acked { count },
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `RequeueInflight` — resurface `agent`'s unacked in-flight messages
|
|
/// (crash recovery on harness boot).
|
|
fn handle_requeue_inflight(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
) -> hive_core_agent_sock::Response {
|
|
match coord.broker.requeue_inflight(agent) {
|
|
Ok(n) => {
|
|
if n > 0 {
|
|
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
|
|
}
|
|
hive_core_agent_sock::Response::Ok
|
|
}
|
|
Err(e) => hive_core_agent_sock::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Unified dispatch for every socket connection — per-agent sockets and the
|
|
/// (now pure-transport) manager socket alike. There is no privilege bit;
|
|
/// authority derives uniformly from the caller's identity: hive-wide
|
|
/// orchestration verbs (schedules) require the matching tool-group (the
|
|
/// grantable capability).
|
|
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 {
|
|
Request::GetLooseEnds => handle_get_loose_ends(coord, agent),
|
|
// Orchestration verbs — gated per-verb on tool-group membership
|
|
// (see `dispatch_orchestration`).
|
|
_ => dispatch_orchestration(req, agent, coord).await,
|
|
}
|
|
}
|
|
|
|
/// Handle the hive-wide orchestration verbs (scheduling). No blanket socket
|
|
/// gate: each verb gates on the grantable capability that authorises it —
|
|
/// the matching tool-group (`scheduling`). Any other variant is a
|
|
/// host-admin / unknown request invalid on either socket.
|
|
async fn dispatch_orchestration(req: &Request, agent: &str, coord: &Arc<Coordinator>) -> Response {
|
|
match req {
|
|
Request::RequestSchedulePrompt(payload) => {
|
|
if let Some(err) = require_group(agent, "scheduling", "schedule a prompt") {
|
|
return err;
|
|
}
|
|
handle_request_schedule_prompt(coord, agent, payload)
|
|
}
|
|
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())
|
|
}
|
|
Request::EditSchedule {
|
|
id,
|
|
body,
|
|
description,
|
|
interval_seconds,
|
|
next_fire_at_unix,
|
|
targets_add,
|
|
targets_remove,
|
|
} => {
|
|
if let Some(err) = require_group(agent, "scheduling", "edit a schedule") {
|
|
return err;
|
|
}
|
|
handle_edit_schedule(
|
|
coord,
|
|
agent,
|
|
*id,
|
|
EditSchedulePatch {
|
|
body: body.clone(),
|
|
description: description.clone(),
|
|
interval_seconds: *interval_seconds,
|
|
next_fire_at_unix: *next_fire_at_unix,
|
|
targets_add: targets_add.clone(),
|
|
targets_remove: targets_remove.clone(),
|
|
},
|
|
)
|
|
}
|
|
Request::ListSchedules => {
|
|
if let Some(err) = require_group(agent, "scheduling", "list schedules") {
|
|
return err;
|
|
}
|
|
handle_list_schedules(coord, agent)
|
|
}
|
|
Request::FireScheduleNow { id } => {
|
|
if let Some(err) = require_group(agent, "scheduling", "fire a schedule") {
|
|
return err;
|
|
}
|
|
handle_fire_schedule_now(coord, agent, *id).await
|
|
}
|
|
// Host-admin-only / unknown variants: never valid on either socket.
|
|
_ => Response::Err {
|
|
message: "request not handled on this socket".to_owned(),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Capability guard for the hive-wide orchestration verbs: the caller must
|
|
/// hold the given tool-group. The tool-group (c0re-owned `tool_groups.json`,
|
|
/// read server-side via [`crate::tool_groups::groups_for`]) is the grantable
|
|
/// 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<Response> {
|
|
if crate::tool_groups::groups_for(agent)
|
|
.iter()
|
|
.any(|g| g == group)
|
|
{
|
|
None
|
|
} else {
|
|
Some(Response::Err {
|
|
message: format!("agent `{agent}` cannot {action}: requires the `{group}` tool group"),
|
|
})
|
|
}
|
|
}
|
|
|
|
/// `GetLooseEnds` — always the caller's own loose ends.
|
|
fn handle_get_loose_ends(coord: &Arc<Coordinator>, agent: &str) -> Response {
|
|
match crate::loose_ends::for_agent(coord, agent) {
|
|
Ok(loose_ends) => Response::LooseEnds { loose_ends },
|
|
Err(e) => Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Handle `CancelLooseEnd` from a per-agent socket. Dispatches by kind, each
|
|
/// with its own auth check: reminder cancels are handled entirely in-container
|
|
/// (this arm should be unreachable in practice, kept only so the match stays
|
|
/// exhaustive) and approval cancels require the `approvals` tool-group (the
|
|
/// grantable capability) AND ownership — the canceller must be the approval's
|
|
/// submitter — so no positional / hardcoded privilege and no cross-agent
|
|
/// cancellation. (The operator's cancel-anything path is a separate handler.)
|
|
fn handle_cancel_loose_end(
|
|
coord: &Arc<Coordinator>,
|
|
canceller: &str,
|
|
kind: hive_sh4re::inbox::CancelLooseEndKind,
|
|
id: i64,
|
|
) -> Result<(), String> {
|
|
match kind {
|
|
hive_sh4re::inbox::CancelLooseEndKind::Reminder => {
|
|
// Reminders are now agent-local (in-container store) — the
|
|
// agent-mcp `cancel_loose_end` tool branches on this kind and
|
|
// dials the agent's own socket directly, never forwarding to
|
|
// hive-c0re. This arm should be unreachable in practice; kept
|
|
// only so the match stays exhaustive.
|
|
Err(format!(
|
|
"reminder {id}: reminders are handled locally by the agent, \
|
|
not by hive-c0re"
|
|
))
|
|
}
|
|
hive_sh4re::inbox::CancelLooseEndKind::Approval => {
|
|
// Withdrawing an approval needs the grantable `approvals`
|
|
// tool-group (held by any approval-submitting orchestrator)
|
|
// AND ownership: only the agent that submitted the approval
|
|
// may withdraw it. Without the ownership check, any
|
|
// approvals-group agent could cancel any other's approval by
|
|
// id. A NULL submitter (legacy row predating the column) is
|
|
// treated as operator-initiated (no agent tracking predates
|
|
// the column).
|
|
check_can_cancel_approval(canceller)?;
|
|
let submitter = coord
|
|
.approvals
|
|
.submitter_of(id)
|
|
.map_err(|e| format!("{e:#}"))?
|
|
.unwrap_or_else(|| "operator".to_owned());
|
|
if submitter != canceller {
|
|
return Err(format!(
|
|
"cancel_loose_end: approval {id} was submitted by {submitter}, \
|
|
not {canceller}; only the submitting agent can withdraw it"
|
|
));
|
|
}
|
|
let approval = coord
|
|
.approvals
|
|
.mark_cancelled(id, canceller)
|
|
.map_err(|e| format!("{e:#}"))?;
|
|
tracing::info!(%id, %canceller, agent = %approval.agent, "approval cancelled");
|
|
let sha_short = approval
|
|
.fetched_sha
|
|
.as_deref()
|
|
.map(|s| s[..s.len().min(12)].to_owned());
|
|
coord.emit_approval_resolved(crate::coordinator::ApprovalResolved {
|
|
id: approval.id,
|
|
agent: approval.agent.as_str(),
|
|
approval_kind: <&str>::from(approval.kind),
|
|
sha_short,
|
|
status: "cancelled",
|
|
note: approval.note,
|
|
description: approval.description,
|
|
});
|
|
Ok(())
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Capability guard on the `Approval` cancel arm: the caller must hold the
|
|
/// `approvals` tool-group (the grantable capability for approval-submitting
|
|
/// orchestrators), checked server-side via `tool_groups::groups_for`. Pulled
|
|
/// out so the auth check has its own focused unit test — exercising the full
|
|
/// `handle_cancel_loose_end` flow would need a `Coordinator` fixture (broker +
|
|
/// sqlite). Keys on a grantable capability, not a positional / hardcoded
|
|
/// privilege.
|
|
fn check_can_cancel_approval(canceller: &str) -> Result<(), String> {
|
|
const APPROVALS_GROUP: &str = "approvals";
|
|
if crate::tool_groups::groups_for(canceller)
|
|
.iter()
|
|
.any(|g| g == APPROVALS_GROUP)
|
|
{
|
|
Ok(())
|
|
} else {
|
|
Err(
|
|
"cancel_loose_end: cancelling approval rows requires the `approvals` tool group"
|
|
.to_owned(),
|
|
)
|
|
}
|
|
}
|
|
|
|
/// Common Send handler shared between dispatch arms. Applies the
|
|
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through
|
|
/// their respective broker calls.
|
|
/// `pub(crate)` so `dispatch_shared` can use it across both socket paths.
|
|
pub(crate) fn handle_send(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
to: &str,
|
|
body: &str,
|
|
in_reply_to: Option<i64>,
|
|
) -> Response {
|
|
if let Err(message) = crate::limits::check_size("send", body) {
|
|
return Response::Err { message };
|
|
}
|
|
if to == "*" {
|
|
let errors = coord.broadcast_send(agent, body);
|
|
return if errors.is_empty() {
|
|
Response::Ok
|
|
} else {
|
|
Response::Err {
|
|
message: format!("broadcast failed for agents: {}", errors.join(", ")),
|
|
}
|
|
};
|
|
}
|
|
// 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.
|
|
//
|
|
// Cross-hive messaging (`name@hive` qualified names) is not routed
|
|
// through the broker — use the Matrix MCP tools for that instead.
|
|
if to.contains('@') {
|
|
return Response::Err {
|
|
message: format!(
|
|
"send failed: cross-hive recipient `{to}` is not supported \
|
|
via the broker — use Matrix MCP tools for cross-hive messaging"
|
|
),
|
|
};
|
|
}
|
|
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)"
|
|
),
|
|
};
|
|
}
|
|
}
|
|
match coord.broker.send(&Message {
|
|
from: hive_sh4re::manager::trusted_sender(agent),
|
|
to: to.to_owned(),
|
|
body: body.to_owned(),
|
|
in_reply_to,
|
|
}) {
|
|
Ok(()) => Response::Ok,
|
|
Err(e) => Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn approval_cancel_rejects_callers_without_the_approvals_group() {
|
|
// A caller that doesn't hold the `approvals` tool-group must not be
|
|
// able to cancel approval rows even if it invents an id. The guard is
|
|
// server-side so client cooperation is irrelevant — and it keys on a
|
|
// grantable capability (the tool-group), not on any agent name.
|
|
// `groups_for` of a name with no tool_groups.json entry is empty.
|
|
let err = check_can_cancel_approval("nobody-with-no-groups").unwrap_err();
|
|
assert!(err.contains("approvals` tool group"), "{err}");
|
|
}
|
|
}
|