23f4f1823b
The page's composer (its input, slash commands and completion), POST /send in hive-agent, the OperatorMsg request of the agent socket and its handler in hive-c0re, and the broker-error helper only the send route used are deleted: writing to an agent is the swarm UI's terminal. cancel turn, new session and log out stay in the state badge's menu; POST /api/compact stays, with nothing on the page calling it. The request-line limit keeps its size, and its comment says it is a generous bound now. The docs describe the actions in place of the footer and the prompt.
657 lines
26 KiB
Rust
657 lines
26 KiB
Rust
//! Unix-socket request server for the per-agent sockets. The socket file's
|
|
//! existence on disk authenticates the caller: connecting to
|
|
//! `<.../agents/foo/mcp.sock>` means you are `foo`. There is no privilege
|
|
//! flag; [`serve`] / [`dispatch`] derive authority uniformly from the caller's
|
|
//! identity. An agent-targeting verb may name any agent.
|
|
|
|
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_sock_client::{LineDeadlines, RequestLine, read_request_line};
|
|
use tokio::io::{AsyncWriteExt, BufReader};
|
|
use tokio::net::{UnixListener, UnixStream};
|
|
use tokio::task::JoinHandle;
|
|
|
|
use crate::coordinator::Coordinator;
|
|
|
|
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 {
|
|
let err = accept_until_fatal(
|
|
&agent,
|
|
|| async { listener.accept().await.map(|(stream, _)| stream) },
|
|
|stream| {
|
|
let agent = agent.clone();
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = serve(stream, agent, coord).await {
|
|
// A peer closing its end mid-read/write (the harness
|
|
// process exiting, a turn ending) is routine and the
|
|
// overwhelming majority of what reaches here —
|
|
// demoted so it doesn't bury a genuine failure.
|
|
if is_broken_pipe(&e) {
|
|
tracing::debug!(error = ?e, "agent connection closed by peer");
|
|
} else {
|
|
tracing::warn!(error = ?e, "agent connection failed");
|
|
}
|
|
}
|
|
});
|
|
},
|
|
)
|
|
.await;
|
|
exit_on_dead_listener(&agent, &err);
|
|
});
|
|
Ok(AgentSocket { path, handle })
|
|
}
|
|
|
|
/// Pause after an accept error that is not about a single connection. EMFILE
|
|
/// and ENFILE leave the listener readable, so retrying at once would spin until
|
|
/// some fd is closed. Same policy as axum's `serve`, which the dashboard runs.
|
|
const ACCEPT_RETRY_DELAY: std::time::Duration = std::time::Duration::from_secs(1);
|
|
|
|
/// Accept errors that mean the listening fd itself is unusable: retrying it
|
|
/// returns the same error forever.
|
|
fn accept_error_is_fatal(e: &std::io::Error) -> bool {
|
|
matches!(
|
|
e.raw_os_error(),
|
|
Some(libc::EBADF | libc::EFAULT | libc::EINVAL | libc::ENOTSOCK | libc::EOPNOTSUPP)
|
|
)
|
|
}
|
|
|
|
/// Hand each accepted connection to `on_conn`, retrying every accept error
|
|
/// except a fatal one, which is returned. `agent` labels the log lines.
|
|
async fn accept_until_fatal<A, F>(
|
|
agent: &str,
|
|
mut accept: A,
|
|
mut on_conn: impl FnMut(UnixStream),
|
|
) -> std::io::Error
|
|
where
|
|
A: FnMut() -> F,
|
|
F: std::future::Future<Output = std::io::Result<UnixStream>>,
|
|
{
|
|
loop {
|
|
match accept().await {
|
|
Ok(stream) => on_conn(stream),
|
|
Err(e) if accept_error_is_fatal(&e) => return e,
|
|
Err(e) => {
|
|
tracing::warn!(%agent, error = ?e, "socket accept failed; retrying");
|
|
if !matches!(
|
|
e.kind(),
|
|
std::io::ErrorKind::ConnectionAborted
|
|
| std::io::ErrorKind::ConnectionReset
|
|
| std::io::ErrorKind::ConnectionRefused
|
|
) {
|
|
tokio::time::sleep(ACCEPT_RETRY_DELAY).await;
|
|
}
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Exit on a listener that can no longer accept. A running hive-c0re re-binds
|
|
/// an agent socket only on that agent's next start; on restart
|
|
/// (`Restart = "on-failure"` on the unit) `mcp_sockets::sync_on_start` re-binds
|
|
/// every one.
|
|
fn exit_on_dead_listener(agent: &str, e: &std::io::Error) -> ! {
|
|
tracing::error!(
|
|
%agent,
|
|
error = ?e,
|
|
"socket listener cannot accept; exiting so hive-c0re restarts and re-binds all sockets"
|
|
);
|
|
std::process::exit(1)
|
|
}
|
|
|
|
/// Whether `e`'s root cause is the peer closing the connection.
|
|
/// `downcast_ref` searches the whole error chain (anyhow's own
|
|
/// `object_downcast`, not just the outermost type), so this finds a
|
|
/// wrapped `std::io::Error` whether or not something added `.context()`
|
|
/// on top of it.
|
|
fn is_broken_pipe(e: &anyhow::Error) -> bool {
|
|
e.downcast_ref::<std::io::Error>().is_some_and(|io_err| {
|
|
matches!(
|
|
io_err.kind(),
|
|
std::io::ErrorKind::BrokenPipe | std::io::ErrorKind::ConnectionReset
|
|
)
|
|
})
|
|
}
|
|
|
|
/// Answer each request line on `stream` until EOF. An over-long (or timed-out) line gets a
|
|
/// `Response::Err` and then the connection is closed, because the rest of the
|
|
/// line is still unread.
|
|
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 = Vec::new();
|
|
loop {
|
|
let outcome = read_request_line(&mut reader, &mut line, LineDeadlines::NONE).await?;
|
|
if outcome == RequestLine::Eof {
|
|
return Ok(());
|
|
}
|
|
let refusal = outcome.refusal();
|
|
let (resp, too_long) = match refusal {
|
|
Some(message) => (Response::Err { message }, true),
|
|
None => match serde_json::from_slice::<Request>(&line) {
|
|
Ok(req) => (dispatch(&req, &agent, &coord).await, false),
|
|
Err(e) => (
|
|
Response::Err {
|
|
message: format!("parse error: {e}"),
|
|
},
|
|
false,
|
|
),
|
|
},
|
|
};
|
|
let mut payload = serde_json::to_string(&resp)?;
|
|
payload.push('\n');
|
|
write.write_all(payload.as_bytes()).await?;
|
|
write.flush().await?;
|
|
if too_long {
|
|
anyhow::bail!("`{agent}` sent an over-long or stalled request line; closed");
|
|
}
|
|
}
|
|
}
|
|
|
|
/// 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,
|
|
}
|
|
}
|
|
|
|
/// Unified dispatch for every socket connection. There is no privilege bit;
|
|
/// authority derives uniformly from the caller's identity.
|
|
async fn dispatch(
|
|
req: &hive_core_agent_sock::Request,
|
|
agent: &str,
|
|
coord: &Arc<Coordinator>,
|
|
) -> hive_core_agent_sock::Response {
|
|
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::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::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
|
|
}
|
|
hive_core_agent_sock::Request::GetLooseEnds => handle_get_loose_ends(coord, agent),
|
|
}
|
|
}
|
|
|
|
/// `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
|
|
}
|
|
|
|
/// `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:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `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:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `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:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Send` handler. Applies the
|
|
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / unicast through
|
|
/// their respective broker calls.
|
|
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. Without this
|
|
// check a typo in `to` silently queues a message nobody will ever read.
|
|
//
|
|
// 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!(
|
|
"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 {
|
|
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),
|
|
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 hive_sock_client::REQUEST_LINE_MAX_BYTES;
|
|
use tokio::io::AsyncBufReadExt;
|
|
|
|
use super::*;
|
|
|
|
/// A real `Coordinator` over a throwaway sqlite dir. `serve` takes an
|
|
/// `Arc<Coordinator>`, so there is no lighter way in; `open` touches
|
|
/// nothing outside the db path it is handed.
|
|
fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
|
|
let dir = tempfile::tempdir().expect("tempdir");
|
|
let coord = Coordinator::open(
|
|
&dir.path().join("broker.sqlite"),
|
|
crate::coordinator::HiveEnv::default(),
|
|
1,
|
|
)
|
|
.expect("open coordinator");
|
|
(dir, Arc::new(coord))
|
|
}
|
|
|
|
#[test]
|
|
fn send_to_operator_is_refused_naming_matrix_and_the_no_matrix_fallback() {
|
|
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__") && message.contains("set_status")),
|
|
"{resp:?}"
|
|
);
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn serve_answers_an_over_long_line_with_an_error_and_closes() {
|
|
let (_dir, coord) = coordinator();
|
|
let (client, server) = UnixStream::pair().expect("socketpair");
|
|
let handler = tokio::spawn(serve(server, "iris".to_owned(), coord));
|
|
let (read, mut write) = client.into_split();
|
|
let writer = tokio::spawn(async move {
|
|
let chunk = vec![b'x'; 64 * 1024];
|
|
for _ in 0..2 * REQUEST_LINE_MAX_BYTES / chunk.len() {
|
|
if write.write_all(&chunk).await.is_err() {
|
|
return;
|
|
}
|
|
}
|
|
let _ = write.shutdown().await;
|
|
});
|
|
let mut reply = String::new();
|
|
BufReader::new(read)
|
|
.read_line(&mut reply)
|
|
.await
|
|
.expect("reply");
|
|
let resp: Response = serde_json::from_str(&reply).expect("response json");
|
|
assert!(
|
|
matches!(&resp, Response::Err { message } if message.starts_with("request too long")),
|
|
"{reply}"
|
|
);
|
|
assert!(
|
|
handler.await.expect("join").is_err(),
|
|
"connection not closed"
|
|
);
|
|
writer.await.expect("join");
|
|
}
|
|
|
|
#[tokio::test]
|
|
async fn accept_loop_survives_a_transient_error() {
|
|
let (client, _peer) = UnixStream::pair().expect("socketpair");
|
|
let mut script = std::collections::VecDeque::from([
|
|
Err(std::io::Error::from_raw_os_error(libc::EMFILE)),
|
|
Ok(client),
|
|
Err(std::io::Error::from_raw_os_error(libc::EBADF)),
|
|
]);
|
|
let mut served = 0;
|
|
let err = accept_until_fatal(
|
|
"iris",
|
|
|| std::future::ready(script.pop_front().expect("accept after a fatal error")),
|
|
|_stream| served += 1,
|
|
)
|
|
.await;
|
|
assert_eq!(served, 1, "connection after EMFILE not served");
|
|
assert_eq!(err.raw_os_error(), Some(libc::EBADF));
|
|
}
|
|
|
|
/// `is_broken_pipe` must recognise both peer-closed kinds whether the
|
|
/// `io::Error` reaches it bare (how `serve`'s own `?`s produce it) or
|
|
/// wrapped in `.context()` (anyhow's `downcast_ref` walks the whole
|
|
/// chain, so wrapping doesn't hide it — confirmed here rather than
|
|
/// assumed), and must not misclassify an unrelated `io::Error` or a
|
|
/// non-io `anyhow::Error`.
|
|
#[test]
|
|
fn is_broken_pipe_matches_peer_closed_kinds_only() {
|
|
let broken_pipe = std::io::Error::from(std::io::ErrorKind::BrokenPipe);
|
|
let conn_reset = std::io::Error::from(std::io::ErrorKind::ConnectionReset);
|
|
let permission_denied = std::io::Error::from(std::io::ErrorKind::PermissionDenied);
|
|
|
|
assert!(is_broken_pipe(&anyhow::Error::new(broken_pipe)));
|
|
assert!(is_broken_pipe(&anyhow::Error::new(conn_reset)));
|
|
assert!(!is_broken_pipe(&anyhow::Error::new(permission_denied)));
|
|
|
|
let contextual = anyhow::Error::new(std::io::Error::from(std::io::ErrorKind::BrokenPipe))
|
|
.context("write response");
|
|
assert!(
|
|
is_broken_pipe(&contextual),
|
|
"anyhow's downcast_ref searches the whole chain, so a \
|
|
.context()-wrapped io::Error must still be found"
|
|
);
|
|
|
|
assert!(!is_broken_pipe(&anyhow::anyhow!("not an io error at all")));
|
|
}
|
|
}
|