refactor(#2455): split hive-agent-mcp into its own bin crate
This commit is contained in:
parent
59dd3a0aa7
commit
e7f8254788
15 changed files with 334 additions and 100 deletions
16
Cargo.lock
generated
16
Cargo.lock
generated
|
|
@ -1489,6 +1489,22 @@ dependencies = [
|
||||||
"url",
|
"url",
|
||||||
]
|
]
|
||||||
|
|
||||||
|
[[package]]
|
||||||
|
name = "hive-agent-mcp"
|
||||||
|
version = "0.1.0"
|
||||||
|
dependencies = [
|
||||||
|
"anyhow",
|
||||||
|
"axum",
|
||||||
|
"clap",
|
||||||
|
"hive-sh4re",
|
||||||
|
"rmcp",
|
||||||
|
"serde",
|
||||||
|
"serde_json",
|
||||||
|
"tokio",
|
||||||
|
"tracing",
|
||||||
|
"tracing-subscriber",
|
||||||
|
]
|
||||||
|
|
||||||
[[package]]
|
[[package]]
|
||||||
name = "hive-agent-wake"
|
name = "hive-agent-wake"
|
||||||
version = "0.1.0"
|
version = "0.1.0"
|
||||||
|
|
|
||||||
|
|
@ -2,6 +2,7 @@
|
||||||
resolver = "3"
|
resolver = "3"
|
||||||
members = [
|
members = [
|
||||||
"hive-ag3nt",
|
"hive-ag3nt",
|
||||||
|
"hive-agent-mcp",
|
||||||
"hive-agent-wake",
|
"hive-agent-wake",
|
||||||
"hive-bash-mcp",
|
"hive-bash-mcp",
|
||||||
"hive-c0re",
|
"hive-c0re",
|
||||||
|
|
|
||||||
|
|
@ -10,9 +10,7 @@ pub mod harness_state;
|
||||||
pub mod identity;
|
pub mod identity;
|
||||||
pub mod login;
|
pub mod login;
|
||||||
pub mod login_session;
|
pub mod login_session;
|
||||||
pub mod mcp;
|
|
||||||
pub mod mcp_config;
|
pub mod mcp_config;
|
||||||
pub mod mcp_loose_ends;
|
|
||||||
pub mod paths;
|
pub mod paths;
|
||||||
pub mod plugins;
|
pub mod plugins;
|
||||||
pub mod prompt;
|
pub mod prompt;
|
||||||
|
|
|
||||||
|
|
@ -2,10 +2,9 @@
|
||||||
//! set into the `--allowedTools` / `--tools` argument strings and renders the
|
//! set into the `--allowedTools` / `--tools` argument strings and renders the
|
||||||
//! `--mcp-config` blob claude reads at spawn (built-in hyperhive server +
|
//! `--mcp-config` blob claude reads at spawn (built-in hyperhive server +
|
||||||
//! any `hyperhive.extraMcpServers`). Pure config-string generation consumed by
|
//! any `hyperhive.extraMcpServers`). Pure config-string generation consumed by
|
||||||
//! [`crate::turn`] when it builds the claude command — it never touches the
|
//! [`crate::turn`] when it builds the claude command. It never touches the
|
||||||
//! running MCP server ([`crate::mcp`]). The `send` allow-list check
|
//! running MCP server (a separate binary) — the `send` allow-list check that
|
||||||
//! ([`check_send_allowed`]) lives here too since it's driven by the same
|
//! server enforces lives alongside it in the `hive-agent-mcp` crate.
|
||||||
//! `/etc/hyperhive/*.json` operator config.
|
|
||||||
|
|
||||||
/// Name of the hyperhive MCP server inside claude's view. Claude prefixes
|
/// Name of the hyperhive MCP server inside claude's view. Claude prefixes
|
||||||
/// tools as `mcp__<this>__<tool>` (e.g. `mcp__hyperhive__send`).
|
/// tools as `mcp__<this>__<tool>` (e.g. `mcp__hyperhive__send`).
|
||||||
|
|
@ -251,59 +250,6 @@ pub fn builtin_tools_arg() -> String {
|
||||||
/// `mcp__<key>__<tool>` pattern in `--allowedTools`.
|
/// `mcp__<key>__<tool>` pattern in `--allowedTools`.
|
||||||
const EXTRA_MCP_PATH: &str = "/etc/hyperhive/extra-mcp.json";
|
const EXTRA_MCP_PATH: &str = "/etc/hyperhive/extra-mcp.json";
|
||||||
|
|
||||||
/// Where the NixOS module writes the per-agent send allow-list (see
|
|
||||||
/// `nix/templates/harness/`). Empty list = unrestricted (the
|
|
||||||
/// default). Non-empty list constrains `mcp__hyperhive__send`'s `to`
|
|
||||||
/// field; the manager is always implicitly permitted regardless of
|
|
||||||
/// the list contents.
|
|
||||||
const SEND_ALLOW_PATH: &str = "/etc/hyperhive/send-allow.json";
|
|
||||||
|
|
||||||
/// Enforce the per-agent send allow-list. Returns `Ok` when the
|
|
||||||
/// recipient is permitted (no list configured, `<parent>` sentinel
|
|
||||||
/// always allowed, or `to` is in the list); returns `Err(refusal)`
|
|
||||||
/// with a claude-readable string when blocked <20><><EFBFBD> the harness surfaces
|
|
||||||
/// the refusal as the tool result so claude knows the message didn't
|
|
||||||
/// land and can react (e.g. route via `<parent>` instead).
|
|
||||||
pub fn check_send_allowed(to: &str) -> Result<(), String> {
|
|
||||||
if to == hive_sh4re::PARENT_RECIPIENT {
|
|
||||||
// Always allow `<parent>` — the allow-list constrains peer
|
|
||||||
// chatter, not the structural reporting line; the operator
|
|
||||||
// can rewire who the parent IS via `set_parent` without
|
|
||||||
// having to remember to update the per-agent allow-list.
|
|
||||||
// The broker resolves the sentinel to the real parent label
|
|
||||||
// on the host side per topology.json (falls back to `operator`
|
|
||||||
// for root agents).
|
|
||||||
return Ok(());
|
|
||||||
}
|
|
||||||
let Ok(raw) = std::fs::read_to_string(SEND_ALLOW_PATH) else {
|
|
||||||
return Ok(()); // file missing → no policy configured → unrestricted
|
|
||||||
};
|
|
||||||
let allow: Vec<String> = match serde_json::from_str(&raw) {
|
|
||||||
Ok(v) => v,
|
|
||||||
Err(e) => {
|
|
||||||
tracing::warn!(
|
|
||||||
path = SEND_ALLOW_PATH,
|
|
||||||
error = ?e,
|
|
||||||
"send allow-list parse failed; falling back to unrestricted",
|
|
||||||
);
|
|
||||||
return Ok(());
|
|
||||||
}
|
|
||||||
};
|
|
||||||
if allow.is_empty() {
|
|
||||||
return Ok(()); // empty list = unrestricted (back-compat)
|
|
||||||
}
|
|
||||||
if allow.iter().any(|n| n == to) {
|
|
||||||
return Ok(());
|
|
||||||
}
|
|
||||||
Err(format!(
|
|
||||||
"send refused: recipient '{to}' not in hyperhive.allowedRecipients \
|
|
||||||
(configured in agent.nix). Allowed: {allow:?}. Your structural \
|
|
||||||
parent is always reachable — route through `send(to: \"{}\", …)` \
|
|
||||||
if you need to reach someone outside the allow-list.",
|
|
||||||
hive_sh4re::PARENT_RECIPIENT
|
|
||||||
))
|
|
||||||
}
|
|
||||||
|
|
||||||
#[derive(Debug, serde::Deserialize)]
|
#[derive(Debug, serde::Deserialize)]
|
||||||
struct ExtraMcpServer {
|
struct ExtraMcpServer {
|
||||||
command: String,
|
command: String,
|
||||||
|
|
|
||||||
|
|
@ -5,7 +5,6 @@
|
||||||
//! stay in the binary because they touch the request enum variants directly.
|
//! stay in the binary because they touch the request enum variants directly.
|
||||||
|
|
||||||
use crate::events::Bus;
|
use crate::events::Bus;
|
||||||
use crate::mcp::REDELIVERY_HINT;
|
|
||||||
use crate::turn::{TurnError, TurnOutcome};
|
use crate::turn::{TurnError, TurnOutcome};
|
||||||
use crate::turn_stats::TurnStatRow;
|
use crate::turn_stats::TurnStatRow;
|
||||||
pub use hive_sh4re::wire_time::now_unix;
|
pub use hive_sh4re::wire_time::now_unix;
|
||||||
|
|
@ -23,38 +22,20 @@ pub fn format_wake_prompt(
|
||||||
unread: u64,
|
unread: u64,
|
||||||
redelivered: bool,
|
redelivered: bool,
|
||||||
) -> String {
|
) -> String {
|
||||||
let banner = if redelivered { REDELIVERY_HINT } else { "" };
|
let banner = if redelivered {
|
||||||
|
hive_sh4re::REDELIVERY_HINT
|
||||||
|
} else {
|
||||||
|
""
|
||||||
|
};
|
||||||
let tag = if id > 0 {
|
let tag = if id > 0 {
|
||||||
format!("[msg #{id}] ")
|
format!("[msg #{id}] ")
|
||||||
} else {
|
} else {
|
||||||
String::new()
|
String::new()
|
||||||
};
|
};
|
||||||
let pending = pending_hint(unread);
|
let pending = hive_sh4re::pending_hint(unread);
|
||||||
format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}")
|
format!("{banner}{tag}Incoming message from `{from}`:\n---\n{body}\n---{pending}")
|
||||||
}
|
}
|
||||||
|
|
||||||
/// Shared "(N more message(s) pending …)" advisory appended after both the
|
|
||||||
/// wake prompt body and the `recv` tool result whenever the inbox still has
|
|
||||||
/// queued messages once the current message/batch is popped. Returns an empty
|
|
||||||
/// string when `remaining == 0`. The leading `\n\n` separates it from the
|
|
||||||
/// preceding body/message block, and the suggested `max` is clamped to the
|
|
||||||
/// server-side recv cap so the hint never asks for more than one round-trip
|
|
||||||
/// can deliver. One builder so the wake prompt and the in-turn recv result
|
|
||||||
/// stay identical.
|
|
||||||
#[must_use]
|
|
||||||
pub fn pending_hint(remaining: u64) -> String {
|
|
||||||
if remaining == 0 {
|
|
||||||
return String::new();
|
|
||||||
}
|
|
||||||
let batch = remaining.min(u64::from(hive_sh4re::RECV_BATCH_MAX));
|
|
||||||
format!(
|
|
||||||
"\n\n({remaining} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \
|
|
||||||
with `max: {batch}` to drain the next batch before acting. If the \
|
|
||||||
backlog is stale/already handled, `ack_until(up_to: <highest [msg #N] seen>)` \
|
|
||||||
clears everything up to that id in one call instead.)"
|
|
||||||
)
|
|
||||||
}
|
|
||||||
|
|
||||||
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
/// Field-named args for [`build_row`]. Mirrors the turn-stats row
|
||||||
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
/// columns; `outcome` and `bus` borrow for the duration of the call.
|
||||||
pub struct TurnRowArgs<'a> {
|
pub struct TurnRowArgs<'a> {
|
||||||
|
|
|
||||||
23
hive-agent-mcp/Cargo.toml
Normal file
23
hive-agent-mcp/Cargo.toml
Normal file
|
|
@ -0,0 +1,23 @@
|
||||||
|
[package]
|
||||||
|
name = "hive-agent-mcp"
|
||||||
|
version.workspace = true
|
||||||
|
edition.workspace = true
|
||||||
|
|
||||||
|
[[bin]]
|
||||||
|
name = "hive-agent-mcp"
|
||||||
|
path = "src/main.rs"
|
||||||
|
|
||||||
|
[lints]
|
||||||
|
workspace = true
|
||||||
|
|
||||||
|
[dependencies]
|
||||||
|
anyhow.workspace = true
|
||||||
|
axum.workspace = true
|
||||||
|
clap.workspace = true
|
||||||
|
hive-sh4re.workspace = true
|
||||||
|
rmcp.workspace = true
|
||||||
|
serde.workspace = true
|
||||||
|
serde_json.workspace = true
|
||||||
|
tokio.workspace = true
|
||||||
|
tracing.workspace = true
|
||||||
|
tracing-subscriber.workspace = true
|
||||||
152
hive-agent-mcp/src/client.rs
Normal file
152
hive-agent-mcp/src/client.rs
Normal file
|
|
@ -0,0 +1,152 @@
|
||||||
|
use std::path::Path;
|
||||||
|
use std::time::Duration;
|
||||||
|
|
||||||
|
use anyhow::{Result, anyhow};
|
||||||
|
use serde::Serialize;
|
||||||
|
use serde::de::DeserializeOwned;
|
||||||
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||||
|
use tokio::net::UnixStream;
|
||||||
|
|
||||||
|
/// Backoff schedule between attempts. Five entries → up to 5 retries on
|
||||||
|
/// top of the initial attempt; total wall-clock cap = 2+4+8+16+30 = 60s.
|
||||||
|
/// Sized to ride out a hive-c0re restart (systemd usually has the unix
|
||||||
|
/// socket back inside ~5s) without the agent-side claude session having
|
||||||
|
/// to handle the transient itself — burning tokens on a tool-error retry
|
||||||
|
/// loop is more expensive than 60s of in-harness sleep.
|
||||||
|
const RETRY_BACKOFFS_MS: &[u64] = &[2_000, 4_000, 8_000, 16_000, 30_000];
|
||||||
|
|
||||||
|
/// Send `req` over the unix socket and decode the single-line JSON
|
||||||
|
/// response, retrying transient connect/IO failures on the backoff
|
||||||
|
/// schedule above and reporting how many retries it took past the initial
|
||||||
|
/// attempt (0 = succeeded first try). MCP tool handlers
|
||||||
|
/// use this so they can append a one-line hint to the tool result when
|
||||||
|
/// retries happened — that way claude knows the prior socket flake
|
||||||
|
/// wasn't a content error and shouldn't trigger an LLM-level retry of
|
||||||
|
/// its own.
|
||||||
|
///
|
||||||
|
/// # Errors
|
||||||
|
///
|
||||||
|
/// Returns an error if all retries are exhausted, or on a fatal protocol
|
||||||
|
/// error (serialization / deserialization failure).
|
||||||
|
///
|
||||||
|
/// # Panics
|
||||||
|
///
|
||||||
|
/// Panics if `RETRY_BACKOFFS_MS.len()` does not fit in a `u32`, which
|
||||||
|
/// cannot happen with the current compile-time constant.
|
||||||
|
pub async fn request_retried<Req, Resp>(socket: &Path, req: &Req) -> Result<(Resp, u32)>
|
||||||
|
where
|
||||||
|
Req: Serialize + ?Sized,
|
||||||
|
Resp: DeserializeOwned,
|
||||||
|
{
|
||||||
|
let mut last_err: Option<anyhow::Error> = None;
|
||||||
|
let max_retries = u32::try_from(RETRY_BACKOFFS_MS.len()).unwrap();
|
||||||
|
for attempt in 0..=max_retries {
|
||||||
|
match try_once::<Req, Resp>(socket, req).await {
|
||||||
|
Ok(resp) => return Ok((resp, attempt)),
|
||||||
|
Err(RequestError::Fatal(e)) => return Err(e),
|
||||||
|
Err(RequestError::Transient(e)) => {
|
||||||
|
if attempt < max_retries {
|
||||||
|
let sleep_ms = RETRY_BACKOFFS_MS[attempt as usize];
|
||||||
|
tracing::warn!(
|
||||||
|
attempt = attempt + 1,
|
||||||
|
sleep_ms,
|
||||||
|
error = %e,
|
||||||
|
"hive socket attempt failed; retrying"
|
||||||
|
);
|
||||||
|
last_err = Some(e);
|
||||||
|
tokio::time::sleep(Duration::from_millis(sleep_ms)).await;
|
||||||
|
} else {
|
||||||
|
last_err = Some(e);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Reaching here means the final attempt returned `Transient`, which always
|
||||||
|
// sets `last_err` — so this is infallible.
|
||||||
|
Err(last_err.expect("a transient failure on the final attempt set last_err"))
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Transient = connect / IO error worth a retry (server restart, broken
|
||||||
|
/// pipe). Fatal = serialization / deserialization / protocol error
|
||||||
|
/// where retrying would just repeat the same failure.
|
||||||
|
enum RequestError {
|
||||||
|
Transient(anyhow::Error),
|
||||||
|
Fatal(anyhow::Error),
|
||||||
|
}
|
||||||
|
|
||||||
|
async fn try_once<Req, Resp>(socket: &Path, req: &Req) -> Result<Resp, RequestError>
|
||||||
|
where
|
||||||
|
Req: Serialize + ?Sized,
|
||||||
|
Resp: DeserializeOwned,
|
||||||
|
{
|
||||||
|
let stream = match UnixStream::connect(socket).await {
|
||||||
|
Ok(stream) => stream,
|
||||||
|
Err(e) => {
|
||||||
|
// A refused or missing socket usually means hive-c0re is
|
||||||
|
// mid-restart (operator redeploy / rebuild) — the socket is
|
||||||
|
// recreated on its boot and `request_retried` rides it out. When
|
||||||
|
// the error *does* surface (retries exhausted, or a non-retried
|
||||||
|
// caller) add that context so claude reads it as a likely
|
||||||
|
// transient rather than a hard failure worth escalating.
|
||||||
|
let restarting = matches!(
|
||||||
|
e.kind(),
|
||||||
|
std::io::ErrorKind::ConnectionRefused | std::io::ErrorKind::NotFound
|
||||||
|
);
|
||||||
|
let mut err = anyhow::Error::new(e).context(format!("connect to {}", socket.display()));
|
||||||
|
if restarting {
|
||||||
|
err = err.context(
|
||||||
|
"hive-c0re may be restarting (e.g. an operator redeploy); \
|
||||||
|
the harness already retried ~60s before surfacing this",
|
||||||
|
);
|
||||||
|
}
|
||||||
|
return Err(RequestError::Transient(err));
|
||||||
|
}
|
||||||
|
};
|
||||||
|
let (read, mut write) = stream.into_split();
|
||||||
|
|
||||||
|
let mut payload = serde_json::to_string(req).map_err(|e| RequestError::Fatal(e.into()))?;
|
||||||
|
payload.push('\n');
|
||||||
|
write
|
||||||
|
.write_all(payload.as_bytes())
|
||||||
|
.await
|
||||||
|
.map_err(|e| RequestError::Transient(e.into()))?;
|
||||||
|
write
|
||||||
|
.flush()
|
||||||
|
.await
|
||||||
|
.map_err(|e| RequestError::Transient(e.into()))?;
|
||||||
|
|
||||||
|
let mut reader = BufReader::new(read);
|
||||||
|
let mut line = String::new();
|
||||||
|
let read_bytes = reader
|
||||||
|
.read_line(&mut line)
|
||||||
|
.await
|
||||||
|
.map_err(|e| RequestError::Transient(e.into()))?;
|
||||||
|
if read_bytes == 0 || line.is_empty() {
|
||||||
|
return Err(RequestError::Transient(anyhow!(
|
||||||
|
"server closed connection without responding"
|
||||||
|
)));
|
||||||
|
}
|
||||||
|
serde_json::from_str(line.trim()).map_err(|e| RequestError::Fatal(e.into()))
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::{RequestError, try_once};
|
||||||
|
|
||||||
|
/// A connect to a non-existent socket path (ENOENT → `NotFound`) is
|
||||||
|
/// classified transient AND annotated with the "hive-c0re is restarting"
|
||||||
|
/// hint, so a surfaced tool error reads as the expected transient.
|
||||||
|
#[tokio::test]
|
||||||
|
async fn missing_socket_connect_is_transient_with_restart_hint() {
|
||||||
|
let bogus = std::path::Path::new("/nonexistent/hive/mcp.sock");
|
||||||
|
match try_once::<(), serde_json::Value>(bogus, &()).await {
|
||||||
|
Err(RequestError::Transient(e)) => {
|
||||||
|
let msg = format!("{e:#}");
|
||||||
|
assert!(msg.contains("restarting"), "missing restart hint: {msg}");
|
||||||
|
assert!(msg.contains("connect to"), "missing connect context: {msg}");
|
||||||
|
}
|
||||||
|
Err(RequestError::Fatal(e)) => panic!("expected transient, got fatal: {e:#}"),
|
||||||
|
Ok(_) => panic!("expected connect failure to a non-existent socket"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
42
hive-agent-mcp/src/loose_ends.rs
Normal file
42
hive-agent-mcp/src/loose_ends.rs
Normal file
|
|
@ -0,0 +1,42 @@
|
||||||
|
//! Generic scanner for MCP loose-end summary files.
|
||||||
|
//!
|
||||||
|
//! External MCP daemons (hive-bash-mcp, hive-matrix-mcp, etc.) write
|
||||||
|
//! JSON files to `$HYPERHIVE_HARNESS_DIR/mcp-loose-ends/<name>.json`.
|
||||||
|
//! Each file contains a JSON array of plain-text summary strings.
|
||||||
|
//!
|
||||||
|
//! The harness reads all files in this directory in `get_loose_ends` to
|
||||||
|
//! surface active background work from any MCP without hardcoding
|
||||||
|
//! per-MCP knowledge here.
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
/// Resolution lives in `hive_sh4re::paths` so the harness + every MCP
|
||||||
|
/// daemon agree on where loose-end summary files are written.
|
||||||
|
fn loose_ends_dir() -> PathBuf {
|
||||||
|
hive_sh4re::paths::mcp_loose_ends_dir()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Collect all loose-end summary strings published by external MCP daemons.
|
||||||
|
/// Each string is a single line suitable for inclusion in `get_loose_ends`
|
||||||
|
/// output. Returns an empty vec if the directory doesn't exist or is empty.
|
||||||
|
#[must_use]
|
||||||
|
pub fn collect() -> Vec<String> {
|
||||||
|
let Ok(rd) = std::fs::read_dir(loose_ends_dir()) else {
|
||||||
|
return Vec::new();
|
||||||
|
};
|
||||||
|
let mut out = Vec::new();
|
||||||
|
for entry in rd.flatten() {
|
||||||
|
let path = entry.path();
|
||||||
|
if path.extension().and_then(|e| e.to_str()) != Some("json") {
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let Ok(content) = std::fs::read_to_string(&path) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
let Ok(items) = serde_json::from_str::<Vec<String>>(&content) else {
|
||||||
|
continue;
|
||||||
|
};
|
||||||
|
out.extend(items);
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
53
hive-agent-mcp/src/main.rs
Normal file
53
hive-agent-mcp/src/main.rs
Normal file
|
|
@ -0,0 +1,53 @@
|
||||||
|
//! MCP-server binary for the built-in hyperhive surface. Runs a long-lived
|
||||||
|
//! streamable-http listener (the `hive-mcp-http` systemd unit) on `--http
|
||||||
|
//! <addr>`; tools dispatch through `/run/hive/mcp.sock` back into the
|
||||||
|
//! hyperhive broker. claude reconnects to the stable URL each turn via
|
||||||
|
//! `--mcp-config`, avoiding the per-turn re-registration race. HTTP is the
|
||||||
|
//! sole transport — there is no stdio mode. Sibling of `hive-agent` (the
|
||||||
|
//! serve loop that renders the `--mcp-config` blob pointing here) and
|
||||||
|
//! `hive-agent-wake`.
|
||||||
|
//!
|
||||||
|
//! Standalone bin crate: the MCP surface (`mcp/`) plus its small support
|
||||||
|
//! modules (socket client, send allow-list, loose-end scanner, path
|
||||||
|
//! resolver) live here rather than in the `hive-ag3nt` harness lib, so the
|
||||||
|
//! server binary doesn't link the whole turn loop.
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
use anyhow::Result;
|
||||||
|
use clap::Parser;
|
||||||
|
|
||||||
|
mod client;
|
||||||
|
mod loose_ends;
|
||||||
|
mod mcp;
|
||||||
|
mod paths;
|
||||||
|
mod send_allow;
|
||||||
|
|
||||||
|
/// Per-agent MCP socket, bind-mounted from the host into every container.
|
||||||
|
const DEFAULT_SOCKET: &str = "/run/hive/mcp.sock";
|
||||||
|
|
||||||
|
#[derive(Parser)]
|
||||||
|
#[command(name = "hive-agent-mcp", about = "hyperhive MCP server")]
|
||||||
|
struct Cli {
|
||||||
|
/// Path to the per-agent MCP socket (bind-mounted from the host).
|
||||||
|
#[arg(long, default_value = DEFAULT_SOCKET)]
|
||||||
|
socket: PathBuf,
|
||||||
|
|
||||||
|
/// Serve streamable-http on this address (e.g. `127.0.0.1:8790`).
|
||||||
|
/// Bind loopback only.
|
||||||
|
#[arg(long)]
|
||||||
|
http: std::net::SocketAddr,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[tokio::main]
|
||||||
|
async fn main() -> Result<()> {
|
||||||
|
tracing_subscriber::fmt()
|
||||||
|
.with_env_filter(
|
||||||
|
tracing_subscriber::EnvFilter::try_from_default_env()
|
||||||
|
.unwrap_or_else(|_| tracing_subscriber::EnvFilter::new("info")),
|
||||||
|
)
|
||||||
|
.init();
|
||||||
|
|
||||||
|
let cli = Cli::parse();
|
||||||
|
mcp::serve_http(cli.socket, cli.http).await
|
||||||
|
}
|
||||||
359
hive-agent-mcp/src/mcp/args.rs
Normal file
359
hive-agent-mcp/src/mcp/args.rs
Normal file
|
|
@ -0,0 +1,359 @@
|
||||||
|
//! Argument structs for the MCP tools: `serde::Deserialize` +
|
||||||
|
//! `schemars::JsonSchema` derives whose field doc-comments become the
|
||||||
|
//! parameter descriptions claude sees in each tool's input schema.
|
||||||
|
|
||||||
|
use rmcp::schemars;
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct SendArgs {
|
||||||
|
/// Logical agent name to deliver the message to (e.g. `"manager"`,
|
||||||
|
/// `"alice"`, or the literal `"operator"` for the dashboard's T4LK box).
|
||||||
|
pub to: String,
|
||||||
|
/// Message body. Plain text; the broker doesn't parse it.
|
||||||
|
pub body: String,
|
||||||
|
/// Optional broker row-id of the message this is a reply to. Lets
|
||||||
|
/// the dashboard render conversation threads. Pass the `id` from the
|
||||||
|
/// `DeliveredMessage` you're responding to; omit for new threads.
|
||||||
|
/// Silently ignored if the id is unknown or out of retention.
|
||||||
|
#[serde(default)]
|
||||||
|
pub in_reply_to: Option<i64>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RecvArgs {
|
||||||
|
/// How long to long-poll for the FIRST message before returning
|
||||||
|
/// the empty marker. Capped at 60s server-side. Default (None)
|
||||||
|
/// is 30s. Useful when an agent wants to park its turn waiting
|
||||||
|
/// for any new work — pick a longer wait to coalesce bursts.
|
||||||
|
#[serde(default)]
|
||||||
|
pub wait_seconds: Option<u64>,
|
||||||
|
/// Maximum number of messages to pop in this round-trip. Default
|
||||||
|
/// (None) is 1 (single-message behaviour — exactly what you want
|
||||||
|
/// when you're called to drive a turn off the first wake). Pass
|
||||||
|
/// a higher value (capped at 5 server-side) when you've been
|
||||||
|
/// told the inbox has more queued (the wake prompt mentions
|
||||||
|
/// pending count) and want to drain everything in one tool call.
|
||||||
|
/// Once the long-poll wakes up, the call drains up to `max` in
|
||||||
|
/// total before returning — no extra round-trip needed.
|
||||||
|
#[serde(default)]
|
||||||
|
pub max: Option<u32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// MCP tool args for `ack_until`.
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct AckUntilArgs {
|
||||||
|
/// Highest broker message id to mark handled: every inbox message
|
||||||
|
/// with `id <= up_to` (ids show as `[msg #<id>]` in recv output)
|
||||||
|
/// is acked in one sweep. Pass the highest id you've actually
|
||||||
|
/// seen/triaged — anything above it stays queued for later turns.
|
||||||
|
pub up_to: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// MCP tool args for `remind`. Exactly one of `delay_seconds` or
|
||||||
|
/// `at_unix_timestamp` must be set; both / neither is a tool-side error.
|
||||||
|
/// Hides the tagged `ReminderTiming` enum behind a flatter schema so the
|
||||||
|
/// model picks one field instead of building `{"timing_type": "in_seconds",
|
||||||
|
/// "seconds": 60}` shaped objects.
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RemindArgs {
|
||||||
|
/// Body that lands in your inbox when the reminder fires (sender
|
||||||
|
/// will appear as `reminder`). Soft cap at 4 KiB inline — anything
|
||||||
|
/// larger gets auto-persisted to a file under
|
||||||
|
/// `/agents/<you>/state/reminders/auto-<ts>.md` and the inbox
|
||||||
|
/// message becomes a short pointer. Pass `file_path` if you want
|
||||||
|
/// to control the destination yourself.
|
||||||
|
pub message: String,
|
||||||
|
/// Fire `delay_seconds` from now (relative). Set this OR
|
||||||
|
/// `at_unix_timestamp`, not both.
|
||||||
|
#[serde(default)]
|
||||||
|
pub delay_seconds: Option<u64>,
|
||||||
|
/// Fire at this absolute unix timestamp (seconds since epoch). Set
|
||||||
|
/// this OR `delay_seconds`, not both.
|
||||||
|
#[serde(default)]
|
||||||
|
pub at_unix_timestamp: Option<i64>,
|
||||||
|
/// Optional path to a file the scheduler should reference instead of
|
||||||
|
/// inlining a long `message`. Use this for large payloads (research
|
||||||
|
/// notes, file lists, intermediate state). Path must be reachable from
|
||||||
|
/// the agent's container — typically under `/agents/<you>/state/`.
|
||||||
|
#[serde(default)]
|
||||||
|
pub file_path: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
// -----------------------------------------------------------------------------
|
||||||
|
// Privileged tool arg types (lifecycle, approvals, scheduling, diagnostics)
|
||||||
|
// -----------------------------------------------------------------------------
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RequestInitConfigArgs {
|
||||||
|
/// New sub-agent name (≤9 chars). Queues an `InitConfig` approval; on
|
||||||
|
/// approval hive-c0re seeds the proposed config repo at
|
||||||
|
/// `/agents/<name>/config/agent.nix` with the default template. After
|
||||||
|
/// the approval the manager edits + commits the config and calls
|
||||||
|
/// `request_apply_commit` to pin the customised sha for the container's
|
||||||
|
/// first build.
|
||||||
|
pub name: String,
|
||||||
|
/// Optional description shown on the dashboard approval card.
|
||||||
|
#[serde(default)]
|
||||||
|
pub description: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct KillArgs {
|
||||||
|
/// Sub-agent name (without the `h-` container prefix).
|
||||||
|
pub name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct SetStatusArgs {
|
||||||
|
/// Status text to display on the dashboard card. Pass an empty string to clear.
|
||||||
|
pub text: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct CreateRepoArgs {
|
||||||
|
/// Repo name — a single segment of letters, digits, `-`, `_`, `.`
|
||||||
|
/// (no leading `-`/`.`). The repo is created as `agents/<repo>`.
|
||||||
|
pub repo: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct GetAgentMetaArgs {
|
||||||
|
/// Logical name of the agent to query (e.g. `"iris"`, `"manager"`).
|
||||||
|
/// Omit to query your own identity + status — replaces the
|
||||||
|
/// previous `whoami` self-introspection tool.
|
||||||
|
#[serde(default)]
|
||||||
|
pub name: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct StartArgs {
|
||||||
|
/// Sub-agent name (without the `h-` container prefix).
|
||||||
|
pub name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RestartArgs {
|
||||||
|
/// Sub-agent name (without the `h-` container prefix).
|
||||||
|
pub name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct UpdateArgs {
|
||||||
|
/// Sub-agent name (without the `h-` container prefix).
|
||||||
|
pub name: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct AskArgs {
|
||||||
|
/// The question to surface.
|
||||||
|
pub question: String,
|
||||||
|
/// Optional fixed-choice answers. The dashboard renders these as
|
||||||
|
/// chips alongside a free-text fallback ("Other…") so the operator
|
||||||
|
/// is never trapped by an incomplete list; peer-agent recipients
|
||||||
|
/// see the list in their inbox event and can return any string.
|
||||||
|
#[serde(default)]
|
||||||
|
pub options: Vec<String>,
|
||||||
|
/// When true, options are rendered as checkboxes — the answerer
|
||||||
|
/// can pick any subset. The answer comes back as a single string
|
||||||
|
/// with selections joined by ", ". Ignored when `options` is empty.
|
||||||
|
#[serde(default)]
|
||||||
|
pub multi: bool,
|
||||||
|
/// Optional auto-cancel after `ttl_seconds` (capped server-side at
|
||||||
|
/// 6 hours). On expiry the question resolves with answer
|
||||||
|
/// `[expired]` and the asker receives the usual
|
||||||
|
/// `question_answered` system event (with `answerer:
|
||||||
|
/// "ttl-watchdog"`). `None` (default) = wait indefinitely.
|
||||||
|
#[serde(default)]
|
||||||
|
pub ttl_seconds: Option<u64>,
|
||||||
|
/// Recipient. Omit (or pass `"operator"`) to ask the human
|
||||||
|
/// operator via the dashboard. Pass another agent's logical name
|
||||||
|
/// to ask that peer — they receive a `question_asked` event in
|
||||||
|
/// their inbox and answer via `mcp__hyperhive__answer`.
|
||||||
|
#[serde(default)]
|
||||||
|
pub to: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct AnswerArgs {
|
||||||
|
/// Id of the question being answered — comes from the
|
||||||
|
/// `question_asked` event in your inbox.
|
||||||
|
pub id: i64,
|
||||||
|
/// Free-text answer body. Soft-capped at 4 KiB by the same
|
||||||
|
/// `MESSAGE_MAX_BYTES` limit as `send`; keep it short or write the
|
||||||
|
/// detail to a file and pass a path.
|
||||||
|
pub answer: String,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct CancelLooseEndArgs {
|
||||||
|
/// Which kind of thread to cancel — `"question"` for an open
|
||||||
|
/// `ask` that's still waiting on an answer, `"reminder"` for a
|
||||||
|
/// scheduled `remind` that hasn't fired yet. Use the `kind`
|
||||||
|
/// field straight off the `get_loose_ends` row.
|
||||||
|
pub kind: String,
|
||||||
|
/// Row id from the matching `get_loose_ends` entry (or the
|
||||||
|
/// `question_queued` reply when you submitted it).
|
||||||
|
pub id: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct AgentGetLooseEndsArgs {
|
||||||
|
/// Whose loose ends to list. Omit (or `null`) for your own. You may
|
||||||
|
/// also pass a direct child agent's name without any extra capability.
|
||||||
|
/// Pass any other agent name to inspect their threads — requires the
|
||||||
|
/// `query_agent_state` capability; without it the request is rejected
|
||||||
|
/// with an error. The `"*"` hive-wide value is not available on the
|
||||||
|
/// agent socket.
|
||||||
|
#[serde(default)]
|
||||||
|
pub agent: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RequestApplyCommitArgs {
|
||||||
|
/// Logical agent name whose config repo the commit lives in.
|
||||||
|
pub agent: String,
|
||||||
|
/// Commit sha (full or short, 7-40 hex chars) in that agent's
|
||||||
|
/// proposed config repo. Must be a sha — a branch or tag name
|
||||||
|
/// (e.g. `main`) is rejected; the approval pins the exact commit.
|
||||||
|
pub commit_ref: String,
|
||||||
|
/// Optional description shown on the dashboard approval card so the
|
||||||
|
/// operator knows what the change does without opening the diff.
|
||||||
|
#[serde(default)]
|
||||||
|
pub description: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct UpdateMetaInputsArgs {
|
||||||
|
/// Flake input names to update (e.g. `["bitburner-agent", "nixpkgs"]`).
|
||||||
|
/// Pass an empty list to update ALL inputs.
|
||||||
|
#[serde(default)]
|
||||||
|
pub inputs: Vec<String>,
|
||||||
|
/// Optional description shown on the dashboard approval card.
|
||||||
|
#[serde(default)]
|
||||||
|
pub description: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct RequestSchedulePromptArgs {
|
||||||
|
/// Recipient agents — one schedule fires to many inboxes at the
|
||||||
|
/// scheduled time. `operator` is a legitimate target (mara: "we
|
||||||
|
/// want to get rid of the manager special case so yes manager
|
||||||
|
/// can be recipient" — the operator slot follows the same rule).
|
||||||
|
pub targets: Vec<String>,
|
||||||
|
/// Message body delivered to each target's inbox at fire time.
|
||||||
|
/// Same size budget as `send` bodies.
|
||||||
|
pub body: String,
|
||||||
|
/// Absolute unix timestamp (seconds) for the FIRST fire. For
|
||||||
|
/// recurring schedules the worker re-arms in
|
||||||
|
/// `interval_seconds` steps from this point on.
|
||||||
|
pub first_fire_at_unix: i64,
|
||||||
|
/// `None` / absent = one-shot. `Some(n > 0)` = recurring every
|
||||||
|
/// `n` seconds. The worker clamps catch-up so a long downtime
|
||||||
|
/// fires ONCE on resume (skipped-cycle count surfaces in the
|
||||||
|
/// per-target `last_result`), not N delayed pulses in a row.
|
||||||
|
#[serde(default)]
|
||||||
|
pub interval_seconds: Option<u64>,
|
||||||
|
/// Optional description shown on the dashboard approval card +
|
||||||
|
/// preserved on the schedule row for later operator reference.
|
||||||
|
#[serde(default)]
|
||||||
|
pub description: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct FireScheduleNowArgs {
|
||||||
|
/// Schedule id to fire out of band. Get this from a prior
|
||||||
|
/// `list_schedules` call or the approval-resolved event for
|
||||||
|
/// the originating `request_schedule_prompt`.
|
||||||
|
pub id: i64,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct CancelScheduleArgs {
|
||||||
|
/// Schedule id from a prior `list_schedules` call or the
|
||||||
|
/// approval-resolved event for a `request_schedule_prompt`.
|
||||||
|
pub id: i64,
|
||||||
|
/// Optional target list. `None` / empty = cancel the entire
|
||||||
|
/// schedule. `Some(["alice", "bob"])` = cancel just those
|
||||||
|
/// recipients (the schedule keeps firing for any remaining
|
||||||
|
/// active targets, and auto-cancels its parent row when every
|
||||||
|
/// target is gone).
|
||||||
|
#[serde(default)]
|
||||||
|
pub targets: Option<Vec<String>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct EditScheduleArgs {
|
||||||
|
/// Schedule id from a prior `list_schedules` call or the
|
||||||
|
/// approval-resolved event for a `request_schedule_prompt`.
|
||||||
|
pub id: i64,
|
||||||
|
/// New body text. Omit to keep the existing one.
|
||||||
|
#[serde(default)]
|
||||||
|
pub body: Option<String>,
|
||||||
|
/// New description. Omit to keep the existing one. (To CLEAR
|
||||||
|
/// the description, use the dashboard PATCH endpoint
|
||||||
|
/// directly — the agent surface intentionally keeps the args
|
||||||
|
/// flat / non-nullable to dodge the doubly-wrapped Option
|
||||||
|
/// schemars quirk; clearing fields is rare and operator-side.)
|
||||||
|
#[serde(default)]
|
||||||
|
pub description: Option<String>,
|
||||||
|
/// Recurring interval in seconds. Omit to keep the existing
|
||||||
|
/// cadence; pass an explicit value to set a new one. Toggling
|
||||||
|
/// recurring↔one-shot (clearing the interval) is operator-only
|
||||||
|
/// for the same reason as `description` above.
|
||||||
|
#[serde(default)]
|
||||||
|
pub interval_seconds: Option<u64>,
|
||||||
|
/// New absolute unix timestamp for the next fire. Omit to
|
||||||
|
/// leave the schedule on its current cadence.
|
||||||
|
#[serde(default)]
|
||||||
|
pub next_fire_at_unix: Option<i64>,
|
||||||
|
/// Names of new targets to add. Replace-on-conflict: re-adding
|
||||||
|
/// a previously cancelled target resets its history (operator
|
||||||
|
/// intent on re-add = "this target is active again").
|
||||||
|
#[serde(default)]
|
||||||
|
pub targets_add: Option<Vec<String>>,
|
||||||
|
/// Names of targets to cancel. Tombstones preserve per-target
|
||||||
|
/// audit; when no active targets remain the schedule
|
||||||
|
/// auto-cancels.
|
||||||
|
#[serde(default)]
|
||||||
|
pub targets_remove: Option<Vec<String>>,
|
||||||
|
}
|
||||||
|
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct GetLogsArgs {
|
||||||
|
/// Logical agent name to fetch logs for (e.g. `gui`, `iris`).
|
||||||
|
/// hive-c0re maps it to the underlying machine name (`h-gui`)
|
||||||
|
/// itself — pass the plain agent name, not the `h-` form.
|
||||||
|
pub agent: String,
|
||||||
|
/// How many journal lines to return (default: 50, max: 500).
|
||||||
|
#[serde(default)]
|
||||||
|
pub lines: Option<u32>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Arguments for `get_host_journal` (capability-gated: `read_host_journal`).
|
||||||
|
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
|
||||||
|
pub struct GetHostJournalArgs {
|
||||||
|
/// Systemd unit to filter (e.g. `hive-c0re.service`). Omit for all units.
|
||||||
|
#[serde(default)]
|
||||||
|
pub unit: Option<String>,
|
||||||
|
/// nspawn machine name verbatim (e.g. `h-iris`). Omit for host journal.
|
||||||
|
/// Agent containers use the `h-` prefix (e.g. `h-iris`); infrastructure
|
||||||
|
/// containers use their full name (e.g. `hive-ci`, `hive-forge`,
|
||||||
|
/// `hive-matrix`, `hive-gateway`).
|
||||||
|
#[serde(default)]
|
||||||
|
pub container: Option<String>,
|
||||||
|
/// Number of lines to return (default 30, max 100).
|
||||||
|
#[serde(default)]
|
||||||
|
pub lines: Option<u32>,
|
||||||
|
/// Minimum syslog priority level.
|
||||||
|
#[serde(default)]
|
||||||
|
pub priority: Option<hive_sh4re::JournalPriority>,
|
||||||
|
/// Regex to match against log message fields (journalctl --grep).
|
||||||
|
#[serde(default)]
|
||||||
|
pub grep: Option<String>,
|
||||||
|
/// Show entries on or newer than this timestamp (e.g. `-1h`).
|
||||||
|
#[serde(default)]
|
||||||
|
pub since: Option<String>,
|
||||||
|
/// Show entries on or older than this timestamp.
|
||||||
|
#[serde(default)]
|
||||||
|
pub until: Option<String>,
|
||||||
|
}
|
||||||
1043
hive-agent-mcp/src/mcp/mod.rs
Normal file
1043
hive-agent-mcp/src/mcp/mod.rs
Normal file
File diff suppressed because it is too large
Load diff
522
hive-agent-mcp/src/mcp/render.rs
Normal file
522
hive-agent-mcp/src/mcp/render.rs
Normal file
|
|
@ -0,0 +1,522 @@
|
||||||
|
//! Formatting / render helpers for the MCP tool surface: ack / recv /
|
||||||
|
//! loose-end / agent-meta reply shaping plus the retry annotation.
|
||||||
|
//! Stateless string builders, with one exception —
|
||||||
|
//! [`matrix_unread_summary`] queries the local matrix daemon socket
|
||||||
|
//! (best-effort) so `get_loose_ends` can prepend an unread-rooms entry.
|
||||||
|
|
||||||
|
/// Render the three identical failure arms every data-returning tool handler
|
||||||
|
/// repeats: a broker `Err` → `"{tool} failed: {m}"`, an unexpected `Ok` variant
|
||||||
|
/// → `"{tool} unexpected response: …"`, and a transport error → `"{tool}
|
||||||
|
/// transport error: …"`. Handlers match their own happy-path variant and route
|
||||||
|
/// everything else here via a catch-all arm (`other => reply_err(other, tool)`),
|
||||||
|
/// so the triplet lives in exactly one place.
|
||||||
|
pub(super) fn reply_err(resp: Result<hive_sh4re::Response, anyhow::Error>, tool: &str) -> String {
|
||||||
|
match resp {
|
||||||
|
Ok(hive_sh4re::Response::Err { message }) => format!("{tool} failed: {message}"),
|
||||||
|
Ok(other) => format!("{tool} unexpected response: {other:?}"),
|
||||||
|
Err(e) => format!("{tool} transport error: {e:#}"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Format helper for "send-like" tools (anything that expects an `Ok`).
|
||||||
|
/// `tool` and `ok_msg` only appear in the result string; they don't change
|
||||||
|
/// behavior.
|
||||||
|
#[must_use]
|
||||||
|
pub fn format_ack(
|
||||||
|
resp: Result<hive_sh4re::Response, anyhow::Error>,
|
||||||
|
tool: &str,
|
||||||
|
ok_msg: String,
|
||||||
|
) -> String {
|
||||||
|
match resp {
|
||||||
|
Ok(hive_sh4re::Response::Ok) => ok_msg,
|
||||||
|
other => reply_err(other, tool),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Format helper for `recv`: renders zero, one, or many popped
|
||||||
|
/// messages. Empty list collapses to "(empty)" so claude doesn't go
|
||||||
|
/// hunting for content; when `waited` is set (the call parked on a
|
||||||
|
/// long-poll that timed out) the empty result also carries
|
||||||
|
/// [`IDLE_WAIT_HINT`] nudging the model toward other work. A single
|
||||||
|
/// message renders as the historical `from: X\n\nbody` block (banner
|
||||||
|
/// first if `redelivered`). A multi-message batch renders with a
|
||||||
|
/// `popped N message(s):` header and `---` separators between bodies
|
||||||
|
/// so the model can tell where one ends and the next begins;
|
||||||
|
/// per-message redelivery banners included.
|
||||||
|
#[must_use]
|
||||||
|
pub fn format_recv(resp: Result<hive_sh4re::Response, anyhow::Error>, waited: bool) -> String {
|
||||||
|
match resp {
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages,
|
||||||
|
remaining,
|
||||||
|
}) => render_recv_messages(&messages, remaining, waited),
|
||||||
|
// A graceful stop is pending — the inbox is fenced. Render a single
|
||||||
|
// explicit directive (not an empty inbox, which claude's "park on recv"
|
||||||
|
// habit would long-poll again, stalling the stop-checkpoint turn until
|
||||||
|
// the drain wait times out into a hard stop) so every recv during the
|
||||||
|
// stop unmissably tells claude to flush + end. `remaining` is forced
|
||||||
|
// to 0 — the inbox is fenced, so a "N more pending" hint would be
|
||||||
|
// misleading.
|
||||||
|
Ok(hive_sh4re::Response::GracefulStop) => {
|
||||||
|
render_recv_messages(&[graceful_stop_message()], 0, waited)
|
||||||
|
}
|
||||||
|
other => reply_err(other, "recv"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// The synthetic single-message directive rendered for a fenced (graceful-stop)
|
||||||
|
/// inbox — see the `GracefulStop` arm of [`format_recv`].
|
||||||
|
fn graceful_stop_message() -> hive_sh4re::DeliveredMessage {
|
||||||
|
hive_sh4re::DeliveredMessage {
|
||||||
|
from: "graceful-stop".into(),
|
||||||
|
body: "⛔ GRACEFUL STOP IN PROGRESS — the container shuts down as soon as this turn \
|
||||||
|
ends. Flush anything worth keeping to your durable /state files, then END \
|
||||||
|
YOUR TURN now. Do NOT call recv again: the inbox is fenced and recv will \
|
||||||
|
only keep returning this same notice."
|
||||||
|
.into(),
|
||||||
|
id: 0,
|
||||||
|
redelivered: false,
|
||||||
|
in_reply_to: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Render the popped-message payload of a successful `recv` (see `format_recv`
|
||||||
|
/// for the empty/single/batch shapes). `remaining` is the post-pop inbox
|
||||||
|
/// depth; when non-zero a shared "(N more pending …)" hint (identical to the
|
||||||
|
/// wake prompt's) is appended so an in-turn drain knows more is queued. The
|
||||||
|
/// empty path never carries the hint (nothing was popped).
|
||||||
|
fn render_recv_messages(
|
||||||
|
messages: &[hive_sh4re::DeliveredMessage],
|
||||||
|
remaining: u64,
|
||||||
|
waited: bool,
|
||||||
|
) -> String {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
if messages.is_empty() {
|
||||||
|
return if waited {
|
||||||
|
format!("(empty){IDLE_WAIT_HINT}")
|
||||||
|
} else {
|
||||||
|
"(empty)".to_owned()
|
||||||
|
};
|
||||||
|
}
|
||||||
|
let mut out = if messages.len() == 1 {
|
||||||
|
let m = &messages[0];
|
||||||
|
let banner = if m.redelivered {
|
||||||
|
hive_sh4re::REDELIVERY_HINT
|
||||||
|
} else {
|
||||||
|
""
|
||||||
|
};
|
||||||
|
format!("{banner}{}from: {}\n\n{}", msg_id_tag(m.id), m.from, m.body)
|
||||||
|
} else {
|
||||||
|
let n = messages.len();
|
||||||
|
let mut out = format!("popped {n} message(s):\n\n");
|
||||||
|
for (i, m) in messages.iter().enumerate() {
|
||||||
|
if i > 0 {
|
||||||
|
out.push_str("\n---\n\n");
|
||||||
|
}
|
||||||
|
let banner = if m.redelivered {
|
||||||
|
hive_sh4re::REDELIVERY_HINT
|
||||||
|
} else {
|
||||||
|
""
|
||||||
|
};
|
||||||
|
let _ = write!(
|
||||||
|
out,
|
||||||
|
"{banner}{}from: {}\n\n{}",
|
||||||
|
msg_id_tag(m.id),
|
||||||
|
m.from,
|
||||||
|
m.body
|
||||||
|
);
|
||||||
|
}
|
||||||
|
out
|
||||||
|
};
|
||||||
|
out.push_str(&hive_sh4re::pending_hint(remaining));
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
/// `[msg #<id>] ` marker prefixed to each recv row so the agent knows
|
||||||
|
/// what to pass to `ack_until` when bulk-triaging a backlog. Transient
|
||||||
|
/// pings carry the sentinel id 0 (in-memory only, nothing in the
|
||||||
|
/// broker to ack) and render without the marker.
|
||||||
|
fn msg_id_tag(id: i64) -> String {
|
||||||
|
if id > 0 {
|
||||||
|
format!("[msg #{id}] ")
|
||||||
|
} else {
|
||||||
|
String::new()
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Appended to the `recv` empty result when the agent parked on a
|
||||||
|
/// long-poll (`wait_seconds > 0`) that timed out with nothing new.
|
||||||
|
/// Nudges the model to spend the idle time on other useful work
|
||||||
|
/// instead of immediately re-blocking on `recv`.
|
||||||
|
pub const IDLE_WAIT_HINT: &str = " — nothing arrived before the wait timed out. \
|
||||||
|
If you have other useful work (assigned issues, in-flight PRs, a docs sweep, \
|
||||||
|
notes to update), do that now rather than immediately parking on recv again.";
|
||||||
|
|
||||||
|
/// Inner renderer for a `Vec<LooseEnd>` already extracted from the socket
|
||||||
|
/// reply. Called by the `get_loose_ends` handler, which injects the
|
||||||
|
/// `UnreadMatrix` entry before formatting.
|
||||||
|
pub(super) fn render_loose_ends(loose_ends: &[hive_sh4re::LooseEnd]) -> String {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
if loose_ends.is_empty() {
|
||||||
|
return "(no loose ends)".to_owned();
|
||||||
|
}
|
||||||
|
let mut out = format!("{} loose end(s):\n", loose_ends.len());
|
||||||
|
for t in loose_ends {
|
||||||
|
match t {
|
||||||
|
hive_sh4re::LooseEnd::Approval {
|
||||||
|
id,
|
||||||
|
agent,
|
||||||
|
commit_ref,
|
||||||
|
description,
|
||||||
|
age_seconds,
|
||||||
|
} => {
|
||||||
|
let desc = description
|
||||||
|
.as_deref()
|
||||||
|
.map(|d| format!(" — {d}"))
|
||||||
|
.unwrap_or_default();
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
"- approval #{id} ({agent} @ {commit_ref}, {age_seconds}s old){desc}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
hive_sh4re::LooseEnd::Question {
|
||||||
|
id,
|
||||||
|
asker,
|
||||||
|
target,
|
||||||
|
question,
|
||||||
|
age_seconds,
|
||||||
|
} => {
|
||||||
|
let to = target.as_deref().unwrap_or("operator");
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
"- question #{id} ({asker} → {to}, {age_seconds}s old): {question}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
hive_sh4re::LooseEnd::Reminder {
|
||||||
|
id,
|
||||||
|
owner,
|
||||||
|
message,
|
||||||
|
due_at,
|
||||||
|
age_seconds,
|
||||||
|
} => {
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
"- reminder #{id} ({owner}, scheduled {age_seconds}s ago, due_at={due_at}): {message}"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
hive_sh4re::LooseEnd::PendingMessages { count } => {
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
"- {count} pending inbox message(s) — drain with recv (recv(max: {count}) to batch)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
hive_sh4re::LooseEnd::UnreadMatrix { rooms, summary } => {
|
||||||
|
let _ = write!(out, "- unread matrix messages in {rooms} room(s)");
|
||||||
|
if summary.is_empty() {
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
" — use list_rooms + read_room to view, mark_read to clear"
|
||||||
|
);
|
||||||
|
} else {
|
||||||
|
let _ = writeln!(out, ":");
|
||||||
|
for line in summary.lines() {
|
||||||
|
let _ = writeln!(out, " {line}");
|
||||||
|
}
|
||||||
|
let _ = writeln!(
|
||||||
|
out,
|
||||||
|
" use list_rooms + read_room to view, mark_read to clear"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Per-room unread entry returned by `matrix_unread_summary`. Mirrors
|
||||||
|
/// `hive-matrix-mcp`'s `RoomUnread` but defined locally to avoid a
|
||||||
|
/// cross-crate dep on the matrix-sdk crate tree.
|
||||||
|
#[derive(Debug, serde::Deserialize)]
|
||||||
|
pub(super) struct MatrixRoomUnread {
|
||||||
|
label: String,
|
||||||
|
count: u32,
|
||||||
|
last_body: Option<String>,
|
||||||
|
last_sender: Option<String>,
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Query the local matrix daemon for per-room unread summaries. Returns
|
||||||
|
/// `None` if the daemon socket is absent or the query fails. Best-effort:
|
||||||
|
/// agents without matrix configured are not penalised.
|
||||||
|
pub(super) async fn matrix_unread_summary() -> Option<Vec<MatrixRoomUnread>> {
|
||||||
|
use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader};
|
||||||
|
use tokio::net::UnixStream;
|
||||||
|
let socket = std::env::var_os("HIVE_MATRIX_SOCKET").map_or_else(
|
||||||
|
|| std::path::PathBuf::from("/run/hive-matrix/socket"),
|
||||||
|
std::path::PathBuf::from,
|
||||||
|
);
|
||||||
|
if !socket.exists() {
|
||||||
|
return None;
|
||||||
|
}
|
||||||
|
let mut stream = UnixStream::connect(&socket).await.ok()?;
|
||||||
|
stream
|
||||||
|
.write_all(b"{\"method\":\"unread_summary\"}\n")
|
||||||
|
.await
|
||||||
|
.ok()?;
|
||||||
|
let mut lines = BufReader::new(stream).lines();
|
||||||
|
let line = lines.next_line().await.ok()??;
|
||||||
|
let val: serde_json::Value = serde_json::from_str(&line).ok()?;
|
||||||
|
// Response: {"kind":"ok","payload":[{label, count, last_body?, last_sender?}]}
|
||||||
|
let arr = val.get("payload")?.as_array()?;
|
||||||
|
serde_json::from_value(serde_json::Value::Array(arr.clone())).ok()
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Format a `Vec<MatrixRoomUnread>` into a per-room summary string.
|
||||||
|
/// Single room / single message collapses to one line; multi-room
|
||||||
|
/// expands to a bulleted list. Returns an empty string for empty input.
|
||||||
|
pub(super) fn format_matrix_summary(rooms: &[MatrixRoomUnread]) -> String {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
if rooms.is_empty() {
|
||||||
|
return String::new();
|
||||||
|
}
|
||||||
|
let mut out = String::new();
|
||||||
|
for r in rooms {
|
||||||
|
if r.count == 1
|
||||||
|
&& let (Some(body), Some(sender)) = (&r.last_body, &r.last_sender)
|
||||||
|
{
|
||||||
|
let _ = writeln!(out, "- {}: {sender}: {body}", r.label);
|
||||||
|
continue;
|
||||||
|
}
|
||||||
|
let _ = writeln!(out, "- {}: {} unread", r.label, r.count);
|
||||||
|
}
|
||||||
|
// Remove trailing newline.
|
||||||
|
if out.ends_with('\n') {
|
||||||
|
out.pop();
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Parse the user-facing `kind` string for `cancel_loose_end` into the
|
||||||
|
/// wire enum. Accepts a small alias set so claude doesn't have to
|
||||||
|
/// remember the exact spelling (`"q"` / `"r"` shorthand falls out
|
||||||
|
/// for free).
|
||||||
|
pub(super) fn parse_loose_end_kind(raw: &str) -> Result<hive_sh4re::CancelLooseEndKind, String> {
|
||||||
|
match raw.trim().to_ascii_lowercase().as_str() {
|
||||||
|
"question" | "q" => Ok(hive_sh4re::CancelLooseEndKind::Question),
|
||||||
|
"reminder" | "r" => Ok(hive_sh4re::CancelLooseEndKind::Reminder),
|
||||||
|
"approval" | "a" => Ok(hive_sh4re::CancelLooseEndKind::Approval),
|
||||||
|
other => Err(format!(
|
||||||
|
"cancel_loose_end: unknown kind '{other}' \
|
||||||
|
(expected \"question\", \"reminder\", or \"approval\")"
|
||||||
|
)),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Canonical user-facing label for a `CancelLooseEndKind` — used in
|
||||||
|
/// the success ack so the caller always sees `"question"` /
|
||||||
|
/// `"reminder"` instead of whatever alias they passed in (`"q"` /
|
||||||
|
/// `"r"`).
|
||||||
|
pub(super) fn loose_end_kind_label(kind: hive_sh4re::CancelLooseEndKind) -> &'static str {
|
||||||
|
match kind {
|
||||||
|
hive_sh4re::CancelLooseEndKind::Question => "question",
|
||||||
|
hive_sh4re::CancelLooseEndKind::Reminder => "reminder",
|
||||||
|
hive_sh4re::CancelLooseEndKind::Approval => "approval",
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Format helper for `get_agent_meta`: renders an agent's identity +
|
||||||
|
/// current status as a short human-readable block. `name`,
|
||||||
|
/// `hyperhive_rev`, and `running` are always shown; `status` only
|
||||||
|
/// appears when one is set, otherwise the line reads `status: <none>`.
|
||||||
|
/// When `running` is false the host has already cleared `status_text`
|
||||||
|
/// (it would be a stale snapshot from before the stop) so the status
|
||||||
|
/// line is implicitly `<none>` in that case — but the explicit
|
||||||
|
/// `running: no` line tells the caller WHY. See
|
||||||
|
/// `docs/turn-loop/mcp.md::Core tools` (`get_agent_meta`).
|
||||||
|
#[must_use]
|
||||||
|
pub fn format_agent_meta(resp: Result<hive_sh4re::Response, anyhow::Error>) -> String {
|
||||||
|
match resp {
|
||||||
|
Ok(hive_sh4re::Response::AgentMeta {
|
||||||
|
name,
|
||||||
|
running,
|
||||||
|
hyperhive_rev,
|
||||||
|
status_text,
|
||||||
|
status_set_at,
|
||||||
|
hive_name,
|
||||||
|
swarm_name,
|
||||||
|
matrix_accounts,
|
||||||
|
}) => {
|
||||||
|
let rev = hyperhive_rev.as_deref().unwrap_or("<unknown>");
|
||||||
|
let run = if running { "yes" } else { "no" };
|
||||||
|
let mut out = format!("name: {name}\nhyperhive_rev: {rev}\nrunning: {run}");
|
||||||
|
// Surface hive + swarm display names only when set, so
|
||||||
|
// single-hive deployments don't see noisy `<none>` lines.
|
||||||
|
if let Some(hn) = hive_name.as_deref() {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
let _ = write!(out, "\nhive_name: {hn}");
|
||||||
|
}
|
||||||
|
if let Some(sn) = swarm_name.as_deref() {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
let _ = write!(out, "\nswarm_name: {sn}");
|
||||||
|
}
|
||||||
|
match status_text {
|
||||||
|
None => out.push_str("\nstatus: <none>"),
|
||||||
|
Some(s) => {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
let age = status_set_at.and_then(|ts| {
|
||||||
|
let now = std::time::SystemTime::now()
|
||||||
|
.duration_since(std::time::UNIX_EPOCH)
|
||||||
|
.ok()?
|
||||||
|
.as_secs();
|
||||||
|
// `ts` is a unix epoch second the agent itself
|
||||||
|
// sourced from `SystemTime` — always positive
|
||||||
|
// in normal operation. Clamp the negative
|
||||||
|
// (clock-skew) edge to 0 before the unsigned
|
||||||
|
// cast so the cast loses no real precision.
|
||||||
|
let ts_secs = u64::try_from(ts).unwrap_or(0);
|
||||||
|
let secs = now.saturating_sub(ts_secs);
|
||||||
|
Some(format_age_secs(secs))
|
||||||
|
});
|
||||||
|
// `write!` into the buffer instead of `push_str(&format!(…))` —
|
||||||
|
// avoids the intermediate allocation clippy::format_push_string
|
||||||
|
// flags. The infallible `String` writer makes this safe to
|
||||||
|
// `let _ =`-ignore.
|
||||||
|
match age {
|
||||||
|
Some(a) => {
|
||||||
|
let _ = write!(out, "\nstatus: {s} (set {a} ago)");
|
||||||
|
}
|
||||||
|
None => {
|
||||||
|
let _ = write!(out, "\nstatus: {s}");
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
}
|
||||||
|
// Matrix identities the agent can act as (the `account` arg on
|
||||||
|
// the matrix tools). Listed only when matrix is provisioned, so
|
||||||
|
// non-matrix agents don't see an empty line.
|
||||||
|
if !matrix_accounts.is_empty() {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
out.push_str("\nmatrix_accounts:");
|
||||||
|
for acct in &matrix_accounts {
|
||||||
|
let uid = acct.user_id.as_deref().unwrap_or("?");
|
||||||
|
let _ = write!(out, "\n {} ({uid}) on {}", acct.name, acct.homeserver);
|
||||||
|
}
|
||||||
|
}
|
||||||
|
out
|
||||||
|
}
|
||||||
|
other => reply_err(other, "get_agent_meta"),
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Format a duration in seconds as a human-readable age string.
|
||||||
|
fn format_age_secs(secs: u64) -> String {
|
||||||
|
if secs < 60 {
|
||||||
|
format!("{secs}s")
|
||||||
|
} else if secs < 3600 {
|
||||||
|
format!("{}m", secs / 60)
|
||||||
|
} else if secs < 86400 {
|
||||||
|
format!("{}h", secs / 3600)
|
||||||
|
} else {
|
||||||
|
format!("{}d", secs / 86400)
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
/// Append a short note to a tool result when the underlying socket call
|
||||||
|
/// took retries to land. Lets claude distinguish "my request was wrong"
|
||||||
|
/// from "c0re flickered and the harness rode it out" — without the
|
||||||
|
/// hint, a tool result that took 30s to come back looks identical to a
|
||||||
|
/// content failure and the model would burn a turn retrying it.
|
||||||
|
#[must_use]
|
||||||
|
pub fn annotate_retries(mut s: String, retries: u32) -> String {
|
||||||
|
if retries > 0 {
|
||||||
|
use std::fmt::Write as _;
|
||||||
|
let suffix = if retries == 1 { "retry" } else { "retries" };
|
||||||
|
let _ = write!(
|
||||||
|
s,
|
||||||
|
"\n\n(note: hive socket connect needed {retries} {suffix} — c0re likely \
|
||||||
|
restarted. Your request did succeed on the final attempt; no action needed.)"
|
||||||
|
);
|
||||||
|
}
|
||||||
|
s
|
||||||
|
}
|
||||||
|
|
||||||
|
#[cfg(test)]
|
||||||
|
mod tests {
|
||||||
|
use super::{IDLE_WAIT_HINT, format_recv};
|
||||||
|
|
||||||
|
fn msg(id: i64, from: &str, body: &str) -> hive_sh4re::DeliveredMessage {
|
||||||
|
hive_sh4re::DeliveredMessage {
|
||||||
|
from: from.to_owned(),
|
||||||
|
body: body.to_owned(),
|
||||||
|
id,
|
||||||
|
redelivered: false,
|
||||||
|
in_reply_to: None,
|
||||||
|
}
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn empty_recv_after_wait_appends_idle_hint() {
|
||||||
|
let out = format_recv(
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![],
|
||||||
|
remaining: 0,
|
||||||
|
}),
|
||||||
|
true,
|
||||||
|
);
|
||||||
|
assert!(out.starts_with("(empty)"));
|
||||||
|
assert!(out.contains(IDLE_WAIT_HINT));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn empty_recv_without_wait_has_no_hint() {
|
||||||
|
let out = format_recv(
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![],
|
||||||
|
remaining: 0,
|
||||||
|
}),
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
assert_eq!(out, "(empty)");
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn single_recv_with_remaining_appends_pending_hint() {
|
||||||
|
let out = format_recv(
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![msg(7, "alice", "hi")],
|
||||||
|
remaining: 3,
|
||||||
|
}),
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
assert!(out.starts_with("[msg #7] from: alice"));
|
||||||
|
assert!(out.contains("3 more message(s) pending"));
|
||||||
|
assert!(out.contains("max: 3"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn single_recv_no_remaining_has_no_pending_hint() {
|
||||||
|
let out = format_recv(
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![msg(7, "alice", "hi")],
|
||||||
|
remaining: 0,
|
||||||
|
}),
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
assert!(!out.contains("more message(s) pending"));
|
||||||
|
}
|
||||||
|
|
||||||
|
#[test]
|
||||||
|
fn batch_recv_with_remaining_appends_pending_hint_once() {
|
||||||
|
let out = format_recv(
|
||||||
|
Ok(hive_sh4re::Response::Messages {
|
||||||
|
messages: vec![msg(7, "alice", "hi"), msg(8, "bob", "yo")],
|
||||||
|
remaining: 9,
|
||||||
|
}),
|
||||||
|
false,
|
||||||
|
);
|
||||||
|
assert!(out.starts_with("popped 2 message(s):"));
|
||||||
|
assert_eq!(out.matches("more message(s) pending").count(), 1);
|
||||||
|
// `max` suggestion is clamped to the server-side recv cap.
|
||||||
|
let batch = 9u64.min(u64::from(hive_sh4re::RECV_BATCH_MAX));
|
||||||
|
assert!(out.contains(&format!("max: {batch}")));
|
||||||
|
}
|
||||||
|
}
|
||||||
19
hive-agent-mcp/src/paths.rs
Normal file
19
hive-agent-mcp/src/paths.rs
Normal file
|
|
@ -0,0 +1,19 @@
|
||||||
|
//! Per-agent path resolution for this MCP server. Self-contained (mirrors
|
||||||
|
//! the sibling helper daemons' own `paths.rs`) so the bin does not link the
|
||||||
|
//! harness lib just to resolve the agent's state dir. Only the one helper
|
||||||
|
//! the MCP server needs lives here.
|
||||||
|
|
||||||
|
use std::path::PathBuf;
|
||||||
|
|
||||||
|
/// Durable state directory for the current agent. Reads `HYPERHIVE_STATE_DIR`
|
||||||
|
/// first (always set by the meta flake to `/agents/{label}/state`); falls back
|
||||||
|
/// to the same pattern derived from `HIVE_LABEL` for dev/test environments
|
||||||
|
/// where the env var may not be set.
|
||||||
|
#[must_use]
|
||||||
|
pub fn state_dir() -> PathBuf {
|
||||||
|
if let Some(p) = std::env::var_os("HYPERHIVE_STATE_DIR") {
|
||||||
|
return PathBuf::from(p);
|
||||||
|
}
|
||||||
|
let label = std::env::var("HIVE_LABEL").unwrap_or_default();
|
||||||
|
PathBuf::from(format!("/agents/{label}/state"))
|
||||||
|
}
|
||||||
58
hive-agent-mcp/src/send_allow.rs
Normal file
58
hive-agent-mcp/src/send_allow.rs
Normal file
|
|
@ -0,0 +1,58 @@
|
||||||
|
//! Per-agent `send` allow-list enforcement. Driven by the operator config at
|
||||||
|
//! `/etc/hyperhive/send-allow.json` (written by the NixOS harness module); the
|
||||||
|
//! MCP server calls [`check_send_allowed`] before forwarding a `send` tool call
|
||||||
|
//! so a blocked recipient surfaces as a claude-readable tool result rather than
|
||||||
|
//! a silent drop.
|
||||||
|
|
||||||
|
/// Where the NixOS module writes the per-agent send allow-list (see
|
||||||
|
/// `nix/templates/harness/`). Empty list = unrestricted (the
|
||||||
|
/// default). Non-empty list constrains `mcp__hyperhive__send`'s `to`
|
||||||
|
/// field; the manager is always implicitly permitted regardless of
|
||||||
|
/// the list contents.
|
||||||
|
const SEND_ALLOW_PATH: &str = "/etc/hyperhive/send-allow.json";
|
||||||
|
|
||||||
|
/// Enforce the per-agent send allow-list. Returns `Ok` when the
|
||||||
|
/// recipient is permitted (no list configured, `<parent>` sentinel
|
||||||
|
/// always allowed, or `to` is in the list); returns `Err(refusal)`
|
||||||
|
/// with a claude-readable string when blocked — the harness surfaces
|
||||||
|
/// the refusal as the tool result so claude knows the message didn't
|
||||||
|
/// land and can react (e.g. route via `<parent>` instead).
|
||||||
|
pub fn check_send_allowed(to: &str) -> Result<(), String> {
|
||||||
|
if to == hive_sh4re::PARENT_RECIPIENT {
|
||||||
|
// Always allow `<parent>` — the allow-list constrains peer
|
||||||
|
// chatter, not the structural reporting line; the operator
|
||||||
|
// can rewire who the parent IS via `set_parent` without
|
||||||
|
// having to remember to update the per-agent allow-list.
|
||||||
|
// The broker resolves the sentinel to the real parent label
|
||||||
|
// on the host side per topology.json (falls back to `operator`
|
||||||
|
// for root agents).
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
let Ok(raw) = std::fs::read_to_string(SEND_ALLOW_PATH) else {
|
||||||
|
return Ok(()); // file missing → no policy configured → unrestricted
|
||||||
|
};
|
||||||
|
let allow: Vec<String> = match serde_json::from_str(&raw) {
|
||||||
|
Ok(v) => v,
|
||||||
|
Err(e) => {
|
||||||
|
tracing::warn!(
|
||||||
|
path = SEND_ALLOW_PATH,
|
||||||
|
error = ?e,
|
||||||
|
"send allow-list parse failed; falling back to unrestricted",
|
||||||
|
);
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
};
|
||||||
|
if allow.is_empty() {
|
||||||
|
return Ok(()); // empty list = unrestricted (back-compat)
|
||||||
|
}
|
||||||
|
if allow.iter().any(|n| n == to) {
|
||||||
|
return Ok(());
|
||||||
|
}
|
||||||
|
Err(format!(
|
||||||
|
"send refused: recipient '{to}' not in hyperhive.allowedRecipients \
|
||||||
|
(configured in agent.nix). Allowed: {allow:?}. Your structural \
|
||||||
|
parent is always reachable — route through `send(to: \"{}\", …)` \
|
||||||
|
if you need to reach someone outside the allow-list.",
|
||||||
|
hive_sh4re::PARENT_RECIPIENT
|
||||||
|
))
|
||||||
|
}
|
||||||
|
|
@ -18,6 +18,34 @@ pub mod wire_time;
|
||||||
/// constant instead of a scattered magic value.
|
/// constant instead of a scattered magic value.
|
||||||
pub const RECV_BATCH_MAX: u32 = 5;
|
pub const RECV_BATCH_MAX: u32 = 5;
|
||||||
|
|
||||||
|
/// Banner prepended to a wake prompt / `recv` result when the message was
|
||||||
|
/// redelivered after a harness restart (the turn that first drove it never
|
||||||
|
/// acked). Shared between the harness serve loop (wake prompt) and the MCP
|
||||||
|
/// server (`recv` tool result) so both surfaces phrase it identically.
|
||||||
|
pub const REDELIVERY_HINT: &str = "[redelivered after harness restart — may already be handled]\n";
|
||||||
|
|
||||||
|
/// Shared "(N more message(s) pending …)" advisory appended after both the
|
||||||
|
/// wake prompt body and the `recv` tool result whenever the inbox still has
|
||||||
|
/// queued messages once the current message/batch is popped. Returns an empty
|
||||||
|
/// string when `remaining == 0`. The leading `\n\n` separates it from the
|
||||||
|
/// preceding body/message block, and the suggested `max` is clamped to the
|
||||||
|
/// server-side recv cap so the hint never asks for more than one round-trip
|
||||||
|
/// can deliver. One builder so the wake prompt (harness serve loop) and the
|
||||||
|
/// in-turn recv result (MCP server) stay identical.
|
||||||
|
#[must_use]
|
||||||
|
pub fn pending_hint(remaining: u64) -> String {
|
||||||
|
if remaining == 0 {
|
||||||
|
return String::new();
|
||||||
|
}
|
||||||
|
let batch = remaining.min(u64::from(RECV_BATCH_MAX));
|
||||||
|
format!(
|
||||||
|
"\n\n({remaining} more message(s) pending in your inbox — call `mcp__hyperhive__recv` \
|
||||||
|
with `max: {batch}` to drain the next batch before acting. If the \
|
||||||
|
backlog is stale/already handled, `ack_until(up_to: <highest [msg #N] seen>)` \
|
||||||
|
clears everything up to that id in one call instead.)"
|
||||||
|
)
|
||||||
|
}
|
||||||
|
|
||||||
/// One row in the approval queue. `commit_ref` is overloaded per
|
/// One row in the approval queue. `commit_ref` is overloaded per
|
||||||
/// `kind` — see `docs/approvals.md::Approval kinds (wire shapes)`
|
/// `kind` — see `docs/approvals.md::Approval kinds (wire shapes)`
|
||||||
/// for the encoding table and lifecycle.
|
/// for the encoding table and lifecycle.
|
||||||
|
|
|
||||||
Loading…
Reference in a new issue