a053d33184
Pure rename ahead of the agent+manager server consolidation: the per-agent socket dispatcher already hosts the shared dispatch and all lifecycle handlers, and will absorb the manager-only handlers next, so `agent_server` becomes a misnomer. No logic change — git mv plus a mechanical `agent_server` -> `socket_server` rename across refs.
1249 lines
47 KiB
Rust
1249 lines
47 KiB
Rust
//! Per-agent socket listener. Each socket file's existence on disk
|
|
//! authenticates the caller: connecting to `<.../agents/foo/mcp.sock>` means
|
|
//! you are `foo`.
|
|
|
|
use std::path::{Path, PathBuf};
|
|
use std::sync::Arc;
|
|
|
|
use anyhow::{Context, Result};
|
|
use hive_sh4re::{AgentRequest, AgentResponse, Message};
|
|
use tokio::io::{AsyncBufReadExt, 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 {
|
|
loop {
|
|
match listener.accept().await {
|
|
Ok((stream, _)) => {
|
|
let agent = agent.clone();
|
|
let coord = coord.clone();
|
|
tokio::spawn(async move {
|
|
if let Err(e) = serve(stream, agent, coord).await {
|
|
tracing::warn!(error = ?e, "agent connection failed");
|
|
}
|
|
});
|
|
}
|
|
Err(e) => {
|
|
tracing::warn!(error = ?e, "agent listener accept failed; exiting");
|
|
return;
|
|
}
|
|
}
|
|
}
|
|
});
|
|
Ok(AgentSocket { path, handle })
|
|
}
|
|
|
|
async fn serve(stream: UnixStream, agent: String, coord: Arc<Coordinator>) -> Result<()> {
|
|
let (read, mut write) = stream.into_split();
|
|
let mut reader = BufReader::new(read);
|
|
let mut line = String::new();
|
|
loop {
|
|
line.clear();
|
|
let n = reader.read_line(&mut line).await?;
|
|
if n == 0 {
|
|
return Ok(());
|
|
}
|
|
let resp = match serde_json::from_str::<AgentRequest>(line.trim()) {
|
|
Ok(req) => dispatch(&req, &agent, &coord).await,
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("parse error: {e}"),
|
|
},
|
|
};
|
|
let mut payload = serde_json::to_string(&resp)?;
|
|
payload.push('\n');
|
|
write.write_all(payload.as_bytes()).await?;
|
|
write.flush().await?;
|
|
}
|
|
}
|
|
|
|
/// Max long-poll window the caller can ask for; values above the
|
|
/// cap are clamped. 180s keeps us under typical TCP/proxy idle
|
|
/// limits while still letting agents park their turn until a
|
|
/// message arrives. Omitting `wait_seconds` (or passing `0`) means
|
|
/// "peek, don't wait" — claude can call recv whenever it wants a
|
|
/// cheap "is there anything pending?" check without blocking the
|
|
/// turn for 30 seconds. To actually park, the caller passes a
|
|
/// positive `wait_seconds`.
|
|
pub(crate) const RECV_LONG_POLL_MAX: std::time::Duration = std::time::Duration::from_mins(3);
|
|
|
|
/// Server-side hard cap on `Recv.max`. Bounds the size of a single
|
|
/// round-trip so a confused caller can't drain the entire inbox in
|
|
/// one go and blow past wire-buffer sizes; everything above the cap
|
|
/// silently clamps. 32 is comfortably above the burst sizes we've
|
|
/// seen in practice (post-rebuild rescue, multi-agent reply storms)
|
|
/// and well under the per-message `MESSAGE_MAX_BYTES` * N envelope
|
|
/// budget.
|
|
pub(crate) const RECV_BATCH_MAX: u32 = 32;
|
|
|
|
pub(crate) fn recv_timeout(wait_seconds: Option<u64>) -> std::time::Duration {
|
|
match wait_seconds {
|
|
Some(s) => std::time::Duration::from_secs(s).min(RECV_LONG_POLL_MAX),
|
|
None => std::time::Duration::ZERO,
|
|
}
|
|
}
|
|
|
|
/// Handle the subset of `Request` variants that are identical on both
|
|
/// the agent socket and the manager socket. Returns `Some(response)` for
|
|
/// every variant it handles; returns `None` for variants with socket-specific
|
|
/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup`
|
|
/// where the manager can target other agents) or for manager-only variants.
|
|
///
|
|
/// Both `socket_server::dispatch` and `manager_server::dispatch` call this
|
|
/// first; each then handles its own remaining arms.
|
|
pub(crate) async fn dispatch_shared(
|
|
req: &hive_sh4re::Request,
|
|
agent: &str,
|
|
privileged: bool,
|
|
coord: &Arc<Coordinator>,
|
|
) -> Option<hive_sh4re::Response> {
|
|
Some(match req {
|
|
hive_sh4re::Request::Send {
|
|
to,
|
|
body,
|
|
in_reply_to,
|
|
} => handle_send(coord, agent, to, body, *in_reply_to),
|
|
hive_sh4re::Request::Recv { wait_seconds, max } => {
|
|
handle_recv(coord, agent, *wait_seconds, *max).await
|
|
}
|
|
hive_sh4re::Request::Status => handle_status(coord, agent),
|
|
hive_sh4re::Request::OperatorMsg { body } => handle_operator_msg(coord, agent, body),
|
|
hive_sh4re::Request::Wake {
|
|
from,
|
|
body,
|
|
transient,
|
|
} => handle_wake(coord, agent, from, body, *transient),
|
|
hive_sh4re::Request::Recent { limit } => handle_recent(coord, agent, *limit),
|
|
hive_sh4re::Request::Ask {
|
|
question,
|
|
options,
|
|
multi,
|
|
ttl_seconds,
|
|
to,
|
|
} => crate::questions::handle_ask(
|
|
coord,
|
|
agent,
|
|
question,
|
|
options,
|
|
*multi,
|
|
*ttl_seconds,
|
|
to.as_deref(),
|
|
)
|
|
.map_or_else(
|
|
|message| hive_sh4re::Response::Err { message },
|
|
|id| hive_sh4re::Response::QuestionQueued { id },
|
|
),
|
|
hive_sh4re::Request::Answer { id, answer } => {
|
|
crate::questions::handle_answer(coord, agent, *id, answer).map_or_else(
|
|
|message| hive_sh4re::Response::Err { message },
|
|
|()| hive_sh4re::Response::Ok,
|
|
)
|
|
}
|
|
hive_sh4re::Request::Remind {
|
|
message,
|
|
timing,
|
|
file_path,
|
|
} => handle_remind(coord, agent, message, timing, file_path.as_deref()),
|
|
hive_sh4re::Request::SetStatus { text } => handle_set_status(coord, text),
|
|
hive_sh4re::Request::GetAgentMeta { name } => {
|
|
handle_get_agent_meta(coord, agent, name.as_deref()).await
|
|
}
|
|
hive_sh4re::Request::CancelLooseEnd { kind, id } => {
|
|
crate::questions::handle_cancel_loose_end(coord, agent, privileged, *kind, *id)
|
|
.map_or_else(
|
|
|message| hive_sh4re::Response::Err { message },
|
|
|()| hive_sh4re::Response::Ok,
|
|
)
|
|
}
|
|
hive_sh4re::Request::CreateRepo { repo } => handle_create_repo(agent, repo).await,
|
|
hive_sh4re::Request::AckTurn => handle_ack_turn(coord, agent),
|
|
hive_sh4re::Request::RequeueInflight => handle_requeue_inflight(coord, agent),
|
|
hive_sh4re::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_sh4re::Response::Ok
|
|
}
|
|
hive_sh4re::Request::GetHostJournal {
|
|
unit,
|
|
container,
|
|
lines,
|
|
priority,
|
|
grep,
|
|
since,
|
|
until,
|
|
} => {
|
|
dispatch_host_journal(
|
|
agent,
|
|
HostJournalArgs {
|
|
unit,
|
|
container,
|
|
lines,
|
|
priority,
|
|
grep,
|
|
since,
|
|
until,
|
|
},
|
|
)
|
|
.await
|
|
}
|
|
// Not a shared variant.
|
|
_ => return None,
|
|
})
|
|
}
|
|
|
|
/// `Recv` — long-poll the broker for up to `max` messages (capped at
|
|
/// `RECV_BATCH_MAX`), mapping deliveries onto the wire response.
|
|
async fn handle_recv(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
wait_seconds: Option<u64>,
|
|
max: Option<u32>,
|
|
) -> hive_sh4re::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_sh4re::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) => hive_sh4re::Response::Messages {
|
|
messages: deliveries
|
|
.into_iter()
|
|
.map(|d| hive_sh4re::DeliveredMessage {
|
|
from: d.message.from,
|
|
body: d.message.body,
|
|
id: d.id,
|
|
redelivered: d.redelivered,
|
|
in_reply_to: d.message.in_reply_to,
|
|
})
|
|
.collect(),
|
|
},
|
|
Err(e) => hive_sh4re::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Wake` — inject a wake into `agent`'s own inbox. Transient wakes
|
|
/// fire the broadcast channel only (no sqlite row, no redelivery on
|
|
/// restart — used by bash-task completions); durable wakes persist
|
|
/// through the broker like any other message.
|
|
fn handle_wake(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
from: &str,
|
|
body: &str,
|
|
transient: bool,
|
|
) -> hive_sh4re::Response {
|
|
let broker = &coord.broker;
|
|
if transient {
|
|
broker.ping(agent, from, body);
|
|
hive_sh4re::Response::Ok
|
|
} else {
|
|
match broker.send(&Message {
|
|
from: from.to_owned(),
|
|
to: agent.to_owned(),
|
|
body: body.to_owned(),
|
|
in_reply_to: None,
|
|
}) {
|
|
Ok(()) => hive_sh4re::Response::Ok,
|
|
Err(e) => hive_sh4re::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_sh4re::Response {
|
|
if let Err(message) = crate::limits::check_status_text(text) {
|
|
return hive_sh4re::Response::Err { message };
|
|
}
|
|
let coord2 = Arc::clone(coord);
|
|
tokio::spawn(async move { coord2.rescan_containers_and_emit().await });
|
|
hive_sh4re::Response::Ok
|
|
}
|
|
|
|
/// Validate an agent-supplied repo name: a single safe slug segment, no
|
|
/// path traversal. Forgejo validates server-side too, but rejecting early
|
|
/// gives a clear message and avoids building odd API paths.
|
|
fn valid_repo_name(name: &str) -> bool {
|
|
!name.is_empty()
|
|
&& name.len() <= 100
|
|
&& !name.starts_with(['-', '.'])
|
|
&& name
|
|
.chars()
|
|
.all(|c| c.is_ascii_alphanumeric() || matches!(c, '-' | '_' | '.'))
|
|
}
|
|
|
|
/// `CreateRepo` — create a repo for `agent` *through hive-c0re* in the
|
|
/// c0re-owned `agents` org with operator-team branch protection.
|
|
/// The sanctioned create path now that agents can't create repos directly.
|
|
async fn handle_create_repo(agent: &str, repo: &str) -> hive_sh4re::Response {
|
|
if !valid_repo_name(repo) {
|
|
return hive_sh4re::Response::Err {
|
|
message: format!(
|
|
"invalid repo name {repo:?} — single segment of letters, digits, '-', '_', '.' \
|
|
(no leading '-'/'.', max 100 chars)"
|
|
),
|
|
};
|
|
}
|
|
let Some(core_token) = crate::forge::core_token() else {
|
|
return hive_sh4re::Response::Err {
|
|
message: "forge unavailable (no core token) — cannot create repo".to_owned(),
|
|
};
|
|
};
|
|
match crate::forge::create_agent_repo(agent, repo, &core_token).await {
|
|
Ok(full_name) => hive_sh4re::Response::RepoCreated {
|
|
clone_url: format!("{}/{full_name}.git", crate::forge::FORGE_HTTP),
|
|
full_name,
|
|
},
|
|
Err(e) => hive_sh4re::Response::Err {
|
|
message: format!("create repo {repo:?} failed: {e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `GetAgentMeta` — identity + live status for `name` (defaults to the
|
|
/// caller). Reads the live container-view status and the hive/swarm
|
|
/// display names.
|
|
async fn handle_get_agent_meta(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
name: Option<&str>,
|
|
) -> hive_sh4re::Response {
|
|
let target = name.unwrap_or(agent);
|
|
let (status_text, status_set_at, running) =
|
|
crate::container_view::read_agent_status_live(target).await;
|
|
let (hive_name, swarm_name) = crate::container_view::hive_swarm_names();
|
|
hive_sh4re::Response::AgentMeta {
|
|
name: target.to_owned(),
|
|
running,
|
|
hyperhive_rev: crate::auto_update::current_flake_rev(&coord.hyperhive_flake),
|
|
status_text,
|
|
status_set_at,
|
|
hive_name,
|
|
swarm_name,
|
|
}
|
|
}
|
|
|
|
/// `Status` — count of pending (unread) inbox messages for `agent`.
|
|
fn handle_status(coord: &Arc<Coordinator>, agent: &str) -> hive_sh4re::Response {
|
|
match coord.broker.count_pending(agent) {
|
|
Ok(unread) => hive_sh4re::Response::Status { unread },
|
|
Err(e) => hive_sh4re::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `OperatorMsg` — deliver an operator-authored message into `agent`'s
|
|
/// inbox (from the `operator` recipient).
|
|
fn handle_operator_msg(coord: &Arc<Coordinator>, agent: &str, body: &str) -> hive_sh4re::Response {
|
|
match coord.broker.send(&Message {
|
|
from: hive_sh4re::OPERATOR_RECIPIENT.to_owned(),
|
|
to: agent.to_owned(),
|
|
body: body.to_owned(),
|
|
in_reply_to: None,
|
|
}) {
|
|
Ok(()) => hive_sh4re::Response::Ok,
|
|
Err(e) => hive_sh4re::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_sh4re::Response {
|
|
match coord.broker.recent_for(agent, limit) {
|
|
Ok(rows) => hive_sh4re::Response::Recent { rows },
|
|
Err(e) => hive_sh4re::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_sh4re::Response {
|
|
match coord.broker.ack_turn(agent) {
|
|
Ok(_n) => hive_sh4re::Response::Ok,
|
|
Err(e) => hive_sh4re::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_sh4re::Response {
|
|
match coord.broker.requeue_inflight(agent) {
|
|
Ok(n) => {
|
|
if n > 0 {
|
|
tracing::info!(%agent, requeued = %n, "requeued in-flight messages");
|
|
}
|
|
hive_sh4re::Response::Ok
|
|
}
|
|
Err(e) => hive_sh4re::Response::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
async fn dispatch(req: &AgentRequest, agent: &str, coord: &Arc<Coordinator>) -> AgentResponse {
|
|
// Regular agent socket: never privileged. Privilege is reserved for
|
|
// requests arriving on the manager socket (see `manager_server`).
|
|
if let Some(resp) = dispatch_shared(req, agent, false, coord).await {
|
|
return resp;
|
|
}
|
|
match req {
|
|
AgentRequest::GetLooseEnds { agent: target } => {
|
|
handle_get_loose_ends(coord, agent, target.as_deref())
|
|
}
|
|
AgentRequest::CountPendingReminders { agent: target } => {
|
|
handle_count_pending_reminders(coord, agent, target.as_deref())
|
|
}
|
|
AgentRequest::ReminderRollup {
|
|
since_secs,
|
|
agent: target,
|
|
} => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs),
|
|
AgentRequest::Start { name } => handle_start_child(coord, agent, name).await,
|
|
AgentRequest::Restart { name } => handle_restart_child(coord, agent, name).await,
|
|
AgentRequest::Kill { name } => handle_kill_child(coord, agent, name).await,
|
|
AgentRequest::Update { name } => handle_update_child(coord, agent, name),
|
|
AgentRequest::ListDescendants => handle_list_descendants(agent).await,
|
|
AgentRequest::RequestInitConfig { name, description } => {
|
|
handle_request_init_config(coord, agent, name, description.clone())
|
|
}
|
|
AgentRequest::RequestApplyCommit {
|
|
agent: target_agent,
|
|
commit_ref,
|
|
description,
|
|
} => {
|
|
handle_request_apply_commit(
|
|
coord,
|
|
agent,
|
|
target_agent,
|
|
commit_ref,
|
|
description.as_deref(),
|
|
)
|
|
.await
|
|
}
|
|
// Manager-only variants are not valid on the agent socket.
|
|
_ => AgentResponse::Err {
|
|
message: "request not supported on agent socket".to_owned(),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Topology guard for the agent-socket lifecycle/config tools: the
|
|
/// caller must be the direct parent of `target`. Returns `Some(Err)`
|
|
/// to short-circuit the dispatch arm when it isn't, `None` when the
|
|
/// call is authorised. `action` is the verb phrase for the message
|
|
/// (e.g. `"start"`, `"request_apply_commit for"`).
|
|
fn require_child(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
|
|
if crate::topology::children_of(agent)
|
|
.iter()
|
|
.any(|c| c == target)
|
|
{
|
|
None
|
|
} else {
|
|
Some(AgentResponse::Err {
|
|
message: format!(
|
|
"agent `{agent}` cannot {action} `{target}`: \
|
|
not a direct child in the topology tree"
|
|
),
|
|
})
|
|
}
|
|
}
|
|
|
|
/// Topology guard for `request_init_config` / `request_apply_commit`,
|
|
/// which may legitimately target a child that does not exist *yet*
|
|
/// (spawning a brand-new sub-agent). The caller may act on a
|
|
/// `target` that is EITHER already its direct child (re-init / config
|
|
/// update of an existing child) OR brand-new (absent from the topology
|
|
/// tree — the requester becomes its parent). A name that already
|
|
/// belongs to a *different* parent (or is a root agent) is refused so
|
|
/// one agent can't hijack another's sub-tree.
|
|
///
|
|
/// Also re-runs the agent-name format check that `require_child`
|
|
/// implicitly provided (a traversal / malformed name could never be a
|
|
/// child): a brand-new name now flows straight to `submit_init_config`,
|
|
/// which builds filesystem paths from it, so validate before that.
|
|
fn require_new_child(agent: &str, target: &str, action: &str) -> Option<AgentResponse> {
|
|
if let Some(reason) = crate::dashboard::validate_agent_name(target) {
|
|
return Some(AgentResponse::Err {
|
|
message: format!("agent `{agent}` cannot {action} `{target}`: {reason}"),
|
|
});
|
|
}
|
|
match crate::topology::read().get(target) {
|
|
// brand-new name — requester becomes the parent on approval.
|
|
None => None,
|
|
// already our direct child — re-init / config update path.
|
|
Some(Some(p)) if p == agent => None,
|
|
// owned by someone else, or a root agent — refuse.
|
|
Some(_) => Some(AgentResponse::Err {
|
|
message: format!(
|
|
"agent `{agent}` cannot {action} `{target}`: it already exists \
|
|
under a different parent in the topology tree"
|
|
),
|
|
}),
|
|
}
|
|
}
|
|
|
|
/// `GetLooseEnds` — resolve the (optionally cross-agent) target then
|
|
/// read its loose ends.
|
|
fn handle_get_loose_ends(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
target: Option<&str>,
|
|
) -> AgentResponse {
|
|
match resolve_agent_state_target(agent, target) {
|
|
Ok(name) => match crate::loose_ends::for_agent(coord, name) {
|
|
Ok(loose_ends) => AgentResponse::LooseEnds { loose_ends },
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
},
|
|
Err(message) => AgentResponse::Err { message },
|
|
}
|
|
}
|
|
|
|
/// `CountPendingReminders` — resolve the target then count its pending
|
|
/// reminders.
|
|
fn handle_count_pending_reminders(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
target: Option<&str>,
|
|
) -> AgentResponse {
|
|
match resolve_agent_state_target(agent, target) {
|
|
Ok(name) => match coord.broker.count_pending_reminders_for(name) {
|
|
Ok(count) => AgentResponse::PendingRemindersCount { count },
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
},
|
|
Err(message) => AgentResponse::Err { message },
|
|
}
|
|
}
|
|
|
|
/// `ReminderRollup` — resolve the target then roll up its reminders
|
|
/// fired in the last `since_secs`.
|
|
fn handle_reminder_rollup(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
target: Option<&str>,
|
|
since_secs: u64,
|
|
) -> AgentResponse {
|
|
match resolve_agent_state_target(agent, target) {
|
|
Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) {
|
|
Ok(stats) => AgentResponse::ReminderRollup(stats),
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
},
|
|
Err(message) => AgentResponse::Err { message },
|
|
}
|
|
}
|
|
|
|
/// `Start` — start a direct-child container, kicking its next turn.
|
|
async fn handle_start_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
|
|
if let Some(err) = require_child(agent, name, "start") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %name, "agent: start child");
|
|
match crate::lifecycle::start(name).await {
|
|
Ok(()) => {
|
|
coord.kick_agent(name, "container started");
|
|
AgentResponse::Ok
|
|
}
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Restart` — enqueue a restart for a direct-child container.
|
|
/// Topology parenthood is the only authorisation criterion — no
|
|
/// capability flag needed.
|
|
async fn handle_restart_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
|
|
// Infra-container restart: an agent holding the `infra_admin`
|
|
// capability can restart a hive infrastructure container (hive-ci /
|
|
// hive-gateway / hive-forge / hive-matrix) by passing its name to the
|
|
// same restart tool. The `InfraContainer` enum parse both recognises
|
|
// these (never agent children, so disjoint from the child path below)
|
|
// and yields the typed value the restart path needs.
|
|
if let Ok(container) = name.parse::<hive_sh4re::priv_proto::InfraContainer>() {
|
|
return handle_restart_infra(coord, agent, container).await;
|
|
}
|
|
if let Some(err) = require_child(agent, name, "restart") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %name, "agent: enqueue restart for child");
|
|
coord.rebuild_queue.enqueue(
|
|
crate::rebuild_queue::QueueKind::Restart,
|
|
name.to_owned(),
|
|
crate::rebuild_queue::QueueSource::Manual,
|
|
format!("agent `{agent}` restart tool"),
|
|
None,
|
|
);
|
|
coord.emit_rebuild_queue_snapshot();
|
|
AgentResponse::Ok
|
|
}
|
|
|
|
/// Restart a hive infrastructure container on behalf of an agent that
|
|
/// holds the `infra_admin` capability. The `container` is already a valid
|
|
/// [`InfraContainer`] (the caller parsed it); this gates on the capability
|
|
/// and routes the systemctl restart through hive-priv. Direct, not
|
|
/// approval-gated.
|
|
async fn handle_restart_infra(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
container: hive_sh4re::priv_proto::InfraContainer,
|
|
) -> AgentResponse {
|
|
let name = container.unit_name();
|
|
// Record the attempt in the operator-visible privileged-action audit
|
|
// trail, then emit a live `AuditEntryAdded` so the dashboard audit view
|
|
// appends it off `/dashboard/stream`. Best-effort: `record` returns the
|
|
// canonical row (or `None` on a sqlite blip), and we stream exactly that
|
|
// row so the stored + streamed views can't drift. `action` is stable so
|
|
// the dashboard can group/filter.
|
|
let audit = |outcome: crate::audit_log::AuditOutcome, detail: Option<&str>| {
|
|
if let Some(entry) = coord
|
|
.audit_log
|
|
.record(agent, "restart_infra", name, outcome, detail)
|
|
{
|
|
coord.emit_audit_entry(entry);
|
|
}
|
|
};
|
|
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::InfraAdmin) {
|
|
tracing::warn!(%agent, %name, "agent: infra restart denied (no infra_admin capability)");
|
|
audit(
|
|
crate::audit_log::AuditOutcome::Err,
|
|
Some("denied: missing infra_admin capability"),
|
|
);
|
|
return AgentResponse::Err {
|
|
message: format!(
|
|
"restarting infra container `{name}` requires the `infra_admin` capability"
|
|
),
|
|
};
|
|
}
|
|
tracing::info!(%agent, %name, "agent: restart infra container");
|
|
match crate::priv_client::restart_infra_container(container).await {
|
|
Ok(()) => {
|
|
audit(crate::audit_log::AuditOutcome::Ok, None);
|
|
AgentResponse::Ok
|
|
}
|
|
Err(e) => {
|
|
let msg = format!("{e:#}");
|
|
audit(crate::audit_log::AuditOutcome::Err, Some(&msg));
|
|
AgentResponse::Err { message: msg }
|
|
}
|
|
}
|
|
}
|
|
|
|
/// `Kill` — kill a direct-child container, unregister it, notify the
|
|
/// manager.
|
|
async fn handle_kill_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
|
|
if let Some(err) = require_child(agent, name, "kill") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %name, "agent: kill child");
|
|
let result: anyhow::Result<()> = async {
|
|
crate::lifecycle::kill(name).await?;
|
|
coord.unregister_agent(name);
|
|
Ok(())
|
|
}
|
|
.await;
|
|
match result {
|
|
Ok(()) => {
|
|
coord.notify_manager(&hive_sh4re::HelperEvent::Killed {
|
|
agent: name.to_owned(),
|
|
});
|
|
AgentResponse::Ok
|
|
}
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `Update` — enqueue a rebuild for a direct-child container.
|
|
fn handle_update_child(coord: &Arc<Coordinator>, agent: &str, name: &str) -> AgentResponse {
|
|
if let Some(err) = require_child(agent, name, "rebuild") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %name, "agent: enqueue rebuild for child");
|
|
coord.rebuild_queue.enqueue(
|
|
crate::rebuild_queue::QueueKind::Rebuild,
|
|
name.to_owned(),
|
|
crate::rebuild_queue::QueueSource::Manual,
|
|
format!("agent `{agent}` update tool"),
|
|
None,
|
|
);
|
|
coord.emit_rebuild_queue_snapshot();
|
|
AgentResponse::Ok
|
|
}
|
|
|
|
/// `ListDescendants` — every topological descendant of `agent` with
|
|
/// its running/stopped state, parents before children.
|
|
async fn handle_list_descendants(agent: &str) -> AgentResponse {
|
|
tracing::debug!(%agent, "agent: list descendants");
|
|
// All containers known to nixos-container (running only).
|
|
let running_set: std::collections::HashSet<String> = match crate::lifecycle::list().await {
|
|
Ok(names) => names
|
|
.into_iter()
|
|
.filter_map(|c| {
|
|
c.strip_prefix(crate::lifecycle::AGENT_PREFIX)
|
|
.map(str::to_owned)
|
|
})
|
|
.collect(),
|
|
Err(e) => {
|
|
return AgentResponse::Err {
|
|
message: format!("list containers failed: {e:#}"),
|
|
};
|
|
}
|
|
};
|
|
// Walk the full topology and collect every descendant.
|
|
let topo = crate::topology::read();
|
|
let mut names: Vec<String> = topo
|
|
.keys()
|
|
.filter(|name| crate::topology::is_descendant_of(name, agent))
|
|
.cloned()
|
|
.collect();
|
|
// Parents before children, then alpha within each tier.
|
|
crate::auto_update::topology_sort(&mut names, &topo);
|
|
let containers = names
|
|
.into_iter()
|
|
.map(|name| {
|
|
let running = running_set.contains(&name);
|
|
hive_sh4re::ContainerInfo { name, running }
|
|
})
|
|
.collect();
|
|
AgentResponse::Containers { containers }
|
|
}
|
|
|
|
/// `RequestInitConfig` — queue an `InitConfig` approval for a
|
|
/// direct-child agent.
|
|
fn handle_request_init_config(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
name: &str,
|
|
description: Option<String>,
|
|
) -> AgentResponse {
|
|
if let Some(err) = require_new_child(agent, name, "request_init_config for") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %name, "agent: request_init_config for child");
|
|
match crate::manager_server::submit_init_config(coord, name, Some(agent), description) {
|
|
Ok(_id) => AgentResponse::Ok,
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// `RequestApplyCommit` — queue an apply-commit approval for a
|
|
/// direct-child agent.
|
|
async fn handle_request_apply_commit(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
target_agent: &str,
|
|
commit_ref: &str,
|
|
description: Option<&str>,
|
|
) -> AgentResponse {
|
|
if let Some(err) = require_new_child(agent, target_agent, "request_apply_commit for") {
|
|
return err;
|
|
}
|
|
tracing::info!(%agent, %target_agent, %commit_ref, "agent: request_apply_commit for child");
|
|
match crate::manager_server::submit_apply_commit(coord, target_agent, commit_ref, description)
|
|
.await
|
|
{
|
|
Ok((id, sha)) => {
|
|
tracing::info!(%id, %target_agent, %sha, "agent: apply_commit approval queued");
|
|
AgentResponse::Ok
|
|
}
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Field-named journal-query knobs for [`dispatch_host_journal`].
|
|
/// Borrows straight from the matched `GetHostJournal` request variant.
|
|
pub struct HostJournalArgs<'a> {
|
|
pub unit: &'a Option<String>,
|
|
pub container: &'a Option<String>,
|
|
pub lines: &'a Option<u32>,
|
|
pub priority: &'a Option<hive_sh4re::JournalPriority>,
|
|
pub grep: &'a Option<String>,
|
|
pub since: &'a Option<String>,
|
|
pub until: &'a Option<String>,
|
|
}
|
|
|
|
/// Handle `GetHostJournal` from both the agent and manager sockets.
|
|
/// Capability-gated: the calling agent must hold `read_host_journal` in
|
|
/// `meta/capabilities.json`. Runs `journalctl` host-side and returns
|
|
/// the output as a `HostJournal` response.
|
|
///
|
|
/// The manager is not exempt - grant `read_host_journal` in
|
|
/// `meta/capabilities.json` to enable it for any agent including the manager.
|
|
pub async fn dispatch_host_journal(agent: &str, args: HostJournalArgs<'_>) -> AgentResponse {
|
|
let HostJournalArgs {
|
|
unit,
|
|
container,
|
|
lines,
|
|
priority,
|
|
grep,
|
|
since,
|
|
until,
|
|
} = args;
|
|
if !crate::capabilities::has_cap(agent, hive_sh4re::Capability::ReadHostJournal) {
|
|
return AgentResponse::Err {
|
|
message: "agent does not have the read_host_journal capability".to_owned(),
|
|
};
|
|
}
|
|
let n = lines.unwrap_or(30).min(100);
|
|
|
|
// A container (`-M`) read enters the container namespace and needs
|
|
// root, so it's delegated to hive-priv. A host read (no container)
|
|
// the unprivileged hive-core user can do directly via its
|
|
// systemd-journal group membership.
|
|
if let Some(c) = container {
|
|
tracing::info!(%agent, machine = %c, %n, "get_host_journal (container)");
|
|
return match crate::priv_client::read_container_journal(
|
|
c,
|
|
hive_sh4re::priv_proto::JournalQuery {
|
|
lines: n,
|
|
unit: unit.clone(),
|
|
priority: priority.as_ref().map(|p| p.as_str().to_owned()),
|
|
grep: grep.clone(),
|
|
since: since.clone(),
|
|
until: until.clone(),
|
|
..Default::default()
|
|
},
|
|
)
|
|
.await
|
|
{
|
|
Ok((stdout, stderr)) => {
|
|
let content = if stdout.is_empty() { stderr } else { stdout };
|
|
AgentResponse::HostJournal { content }
|
|
}
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("journal read: {e:#}"),
|
|
},
|
|
};
|
|
}
|
|
|
|
let mut args: Vec<String> = vec![
|
|
"--no-pager".to_owned(),
|
|
"--output=short".to_owned(),
|
|
"-n".to_owned(),
|
|
n.to_string(),
|
|
];
|
|
if let Some(u) = unit {
|
|
args.push("-u".to_owned());
|
|
args.push(u.clone());
|
|
}
|
|
if let Some(p) = priority {
|
|
args.push("-p".to_owned());
|
|
args.push(p.as_str().to_owned());
|
|
}
|
|
if let Some(g) = grep {
|
|
args.push(format!("--grep={g}"));
|
|
}
|
|
if let Some(s) = since {
|
|
args.push(format!("--since={s}"));
|
|
}
|
|
if let Some(u) = until {
|
|
args.push(format!("--until={u}"));
|
|
}
|
|
tracing::info!(%agent, ?args, "get_host_journal");
|
|
match tokio::process::Command::new("journalctl")
|
|
.args(&args)
|
|
.output()
|
|
.await
|
|
{
|
|
Ok(out) => {
|
|
let content = if out.status.success() || !out.stdout.is_empty() {
|
|
String::from_utf8_lossy(&out.stdout).into_owned()
|
|
} else {
|
|
let stderr = String::from_utf8_lossy(&out.stderr);
|
|
format!("journalctl exited {}: {stderr}", out.status)
|
|
};
|
|
AgentResponse::HostJournal { content }
|
|
}
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("journalctl spawn failed: {e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
/// Fan out one message to each recipient in `targets`. Skips the sender
|
|
/// itself. Returns a list of `"<agent>: <error>"` strings for any delivery
|
|
/// failures (empty = all good).
|
|
pub(crate) fn fan_out_send(
|
|
coord: &Arc<Coordinator>,
|
|
from: &str,
|
|
body: &str,
|
|
in_reply_to: Option<i64>,
|
|
targets: &[String],
|
|
) -> Vec<String> {
|
|
let mut errors = Vec::new();
|
|
for target in targets {
|
|
if target == from {
|
|
continue;
|
|
}
|
|
if let Err(e) = coord.broker.send(&Message {
|
|
from: from.to_owned(),
|
|
to: target.clone(),
|
|
body: body.to_owned(),
|
|
in_reply_to,
|
|
}) {
|
|
errors.push(format!("{target}: {e}"));
|
|
}
|
|
}
|
|
errors
|
|
}
|
|
|
|
/// Common Send handler shared between dispatch arms. Applies the
|
|
/// 4 KiB body cap, then routes broadcast (`to == "*"`) / children fan-out
|
|
/// (`to == "<children>"`) / unicast through their respective broker calls.
|
|
/// `pub(crate)` so `dispatch_shared` (and via it, `manager_server`) can use it.
|
|
pub(crate) fn handle_send(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
to: &str,
|
|
body: &str,
|
|
in_reply_to: Option<i64>,
|
|
) -> AgentResponse {
|
|
if let Err(message) = crate::limits::check_size("send", body) {
|
|
return AgentResponse::Err { message };
|
|
}
|
|
if to == "*" {
|
|
let errors = coord.broadcast_send(agent, body);
|
|
return if errors.is_empty() {
|
|
AgentResponse::Ok
|
|
} else {
|
|
AgentResponse::Err {
|
|
message: format!("broadcast failed for agents: {}", errors.join(", ")),
|
|
}
|
|
};
|
|
}
|
|
// `<children>`: fan out to every direct descendant of the sender per
|
|
// topology.json. Bypasses the allow-list check — structural fan-out
|
|
// targets are never user-listed peers. No-op (returns Ok) for leaf
|
|
// agents that have no children.
|
|
if to == hive_sh4re::CHILDREN_RECIPIENT {
|
|
let children = crate::topology::children_of(agent);
|
|
let errors = fan_out_send(coord, agent, body, in_reply_to, &children);
|
|
return if errors.is_empty() {
|
|
AgentResponse::Ok
|
|
} else {
|
|
AgentResponse::Err {
|
|
message: format!("children fan-out failed for agents: {}", errors.join(", ")),
|
|
}
|
|
};
|
|
}
|
|
// Resolve magic-recipient sentinels (`<parent>`) against topology.json;
|
|
// no-op for ordinary names. Lets agents address structural roles without
|
|
// learning the label — runtime reparenting propagates for free. See
|
|
// `docs/conventions.md::Recipient sentinels`.
|
|
let resolved = crate::topology::resolve_recipient(agent, to);
|
|
// Validate that the resolved recipient is a known local agent or the
|
|
// special "operator" recipient. Without this check a typo in `to`
|
|
// silently queues a message nobody will ever read.
|
|
//
|
|
// Cross-hive messaging (`name@hive` qualified names) is not routed
|
|
// through the broker — use the Matrix MCP tools for that instead.
|
|
if resolved.contains('@') {
|
|
return AgentResponse::Err {
|
|
message: format!(
|
|
"send failed: cross-hive recipient `{resolved}` is not supported \
|
|
via the broker — use Matrix MCP tools for cross-hive messaging"
|
|
),
|
|
};
|
|
}
|
|
if resolved != hive_sh4re::OPERATOR_RECIPIENT {
|
|
let state_root = crate::coordinator::Coordinator::agent_state_root(&resolved);
|
|
if !state_root.exists() {
|
|
return AgentResponse::Err {
|
|
message: format!(
|
|
"send failed: unknown recipient `{resolved}` \
|
|
(no agent with that name exists on this hive)"
|
|
),
|
|
};
|
|
}
|
|
}
|
|
match coord.broker.send(&Message {
|
|
from: agent.to_owned(),
|
|
to: resolved,
|
|
body: body.to_owned(),
|
|
in_reply_to,
|
|
}) {
|
|
Ok(()) => AgentResponse::Ok,
|
|
Err(e) => AgentResponse::Err {
|
|
message: format!("{e:#}"),
|
|
},
|
|
}
|
|
}
|
|
|
|
fn handle_remind(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
message: &str,
|
|
timing: &hive_sh4re::ReminderTiming,
|
|
file_path: Option<&str>,
|
|
) -> AgentResponse {
|
|
match store_remind(coord, agent, message, timing, file_path) {
|
|
Ok(()) => AgentResponse::Ok,
|
|
Err(message) => AgentResponse::Err { message },
|
|
}
|
|
}
|
|
|
|
/// Shared remind-storage path used by both the agent and the manager
|
|
/// dispatchers. Validates timing, applies the auto-file overflow
|
|
/// dance (see [`prepare_remind_storage`]), and writes the reminder
|
|
/// row. Returns `Ok(())` on success, or a caller-ready error string
|
|
/// the dispatcher wraps in `*Response::Err`.
|
|
/// Maximum pending (un-delivered) reminders per agent. Exceeding this
|
|
/// causes `store_remind` to return an error so the agent knows to back
|
|
/// off instead of silently dropping. Override via
|
|
/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; set to `0` to disable the cap
|
|
/// (not recommended — a runaway agent can still flood the scheduler).
|
|
const DEFAULT_REMIND_MAX_PENDING: u64 = 50;
|
|
|
|
fn remind_max_pending() -> u64 {
|
|
std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT")
|
|
.ok()
|
|
.and_then(|s| s.trim().parse::<u64>().ok())
|
|
.unwrap_or(DEFAULT_REMIND_MAX_PENDING)
|
|
}
|
|
|
|
pub(crate) fn store_remind(
|
|
coord: &Arc<Coordinator>,
|
|
agent: &str,
|
|
message: &str,
|
|
timing: &hive_sh4re::ReminderTiming,
|
|
file_path: Option<&str>,
|
|
) -> Result<(), String> {
|
|
let max = remind_max_pending();
|
|
if max > 0 {
|
|
let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0);
|
|
if pending >= max {
|
|
return Err(format!(
|
|
"reminder rejected: agent `{agent}` already has {pending} pending \
|
|
reminders (cap {max}). Cancel some via `cancel_loose_end` or wait \
|
|
for them to fire before scheduling more. Override the cap with \
|
|
`HIVE_REMIND_MAX_PENDING_PER_AGENT`."
|
|
));
|
|
}
|
|
}
|
|
let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?;
|
|
let (stored_message, stored_path) = prepare_remind_storage(agent, message, file_path)?;
|
|
let id = coord
|
|
.broker
|
|
.store_reminder(agent, &stored_message, stored_path.as_deref(), due_at)
|
|
.map_err(|e| format!("failed to store reminder: {e:#}"))?;
|
|
tracing::info!(%id, %agent, %due_at, "reminder scheduled");
|
|
coord.emit_reminders_snapshot();
|
|
Ok(())
|
|
}
|
|
|
|
/// Decide what we actually store in the reminders row, applying the
|
|
/// same byte cap as the rest of the wire protocol
|
|
/// ([`crate::limits::MESSAGE_MAX_BYTES`]). Three outcomes:
|
|
///
|
|
/// 1. Body within the cap → stored verbatim, with whatever `file_path`
|
|
/// the caller passed (None or Some). The scheduler honours
|
|
/// `file_path` at delivery time as before.
|
|
/// 2. Body over the cap, no caller `file_path` → auto-generate a path
|
|
/// under `/agents/<agent>/state/reminders/auto-<ts>.md`, write the
|
|
/// body to disk now, store a short pointer hint as the message and
|
|
/// clear `file_path` (so the scheduler doesn't re-write at
|
|
/// delivery and overwrite the body with the hint).
|
|
/// 3. Body over the cap, caller provided `file_path` → honour the
|
|
/// caller's path: write the body to it now, store the same hint
|
|
/// and clear `file_path` for the same reason as (2).
|
|
///
|
|
/// Returns `(stored_message, stored_file_path)` on success, or a
|
|
/// caller-ready error string on auto-save failure (which is the only
|
|
/// way a Remind request can be refused for size — the agent never has
|
|
/// to think about the cap).
|
|
fn prepare_remind_storage(
|
|
agent: &str,
|
|
message: &str,
|
|
file_path: Option<&str>,
|
|
) -> Result<(String, Option<String>), String> {
|
|
if message.len() <= crate::limits::MESSAGE_MAX_BYTES {
|
|
return Ok((message.to_owned(), file_path.map(str::to_owned)));
|
|
}
|
|
let req_path = match file_path {
|
|
Some(p) => p.to_owned(),
|
|
None => auto_reminder_path(agent),
|
|
};
|
|
let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path)
|
|
.map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?;
|
|
crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| {
|
|
format!("auto-save of large reminder body to `{req_path}` failed: {reason}")
|
|
})?;
|
|
let hint = format!(
|
|
"[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]",
|
|
message.len()
|
|
);
|
|
Ok((hint, None))
|
|
}
|
|
|
|
/// Generate a per-agent path for an auto-saved reminder body. Uses
|
|
/// `unix_nanos` plus the agent name to keep collisions infinitesimal
|
|
/// across the agent's own state subtree (we're not stamping a hostname
|
|
/// since hive-c0re is single-host).
|
|
fn auto_reminder_path(agent: &str) -> String {
|
|
let ts_ns = std::time::SystemTime::now()
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map_or(0, |d| d.as_nanos());
|
|
format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md")
|
|
}
|
|
|
|
/// Resolve the target agent name for `GetLooseEnds`, `CountPendingReminders`,
|
|
/// and `ReminderRollup` on the agent socket. Rules:
|
|
///
|
|
/// - `None` → caller's own threads (always allowed).
|
|
/// - `Some(caller)` → same as `None`.
|
|
/// - `Some("<child>")` where child is a direct descendant of caller per
|
|
/// `topology.json` → allowed without any extra capability.
|
|
/// - `Some("<other>")` where other is not a child → requires the
|
|
/// `query_agent_state` capability; returns an error otherwise.
|
|
/// - `Some("*")` → always rejected (hive-wide scans are manager-only).
|
|
fn resolve_agent_state_target<'a>(
|
|
caller: &'a str,
|
|
target: Option<&'a str>,
|
|
) -> Result<&'a str, String> {
|
|
match target {
|
|
None => Ok(caller),
|
|
Some("*") => Err(
|
|
"hive-wide query (agent=\"*\") is not available on the agent socket; \
|
|
use the manager socket for swarm-wide scans"
|
|
.to_owned(),
|
|
),
|
|
Some(name) => {
|
|
if name == caller {
|
|
return Ok(caller);
|
|
}
|
|
// Direct children are visible to their parent without extra capability.
|
|
if crate::topology::children_of(caller)
|
|
.iter()
|
|
.any(|c| c == name)
|
|
{
|
|
return Ok(name);
|
|
}
|
|
if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) {
|
|
Ok(name)
|
|
} else {
|
|
Err(format!(
|
|
"agent `{caller}` cannot query `{name}`: not a direct child and \
|
|
`query_agent_state` capability is not granted"
|
|
))
|
|
}
|
|
}
|
|
}
|
|
}
|
|
|
|
/// Resolve the `due_at` unix timestamp for a Remind request. Returns
|
|
/// distinct error messages for each failure mode (overflow on
|
|
/// `InSeconds`, pre-epoch clock, `i64` cast wrap) so the caller can tell
|
|
/// what went wrong without inspecting the chain.
|
|
fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result<i64> {
|
|
use hive_sh4re::ReminderTiming;
|
|
match timing {
|
|
ReminderTiming::InSeconds { seconds } => {
|
|
let now = std::time::SystemTime::now();
|
|
let future = now
|
|
.checked_add(std::time::Duration::from_secs(*seconds))
|
|
.ok_or_else(|| {
|
|
anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range")
|
|
})?;
|
|
let duration = future
|
|
.duration_since(std::time::UNIX_EPOCH)
|
|
.map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?;
|
|
i64::try_from(duration.as_secs())
|
|
.map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}"))
|
|
}
|
|
ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp),
|
|
}
|
|
}
|
|
|
|
#[cfg(test)]
|
|
mod tests {
|
|
use super::*;
|
|
|
|
#[test]
|
|
fn auto_reminder_path_format() {
|
|
let p = auto_reminder_path("damocles");
|
|
assert!(p.starts_with("/agents/damocles/state/reminders/auto-"));
|
|
assert!(
|
|
std::path::Path::new(&p)
|
|
.extension()
|
|
.is_some_and(|ext| ext.eq_ignore_ascii_case("md"))
|
|
);
|
|
}
|
|
|
|
#[test]
|
|
fn prepare_remind_storage_passthrough_under_cap() {
|
|
let (msg, fp) = prepare_remind_storage("foo", "small body", None).unwrap();
|
|
assert_eq!(msg, "small body");
|
|
assert_eq!(fp, None);
|
|
}
|
|
|
|
#[test]
|
|
fn prepare_remind_storage_passthrough_with_caller_file_path() {
|
|
let (msg, fp) =
|
|
prepare_remind_storage("foo", "small", Some("/agents/foo/state/x.md")).unwrap();
|
|
assert_eq!(msg, "small");
|
|
assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md"));
|
|
}
|
|
}
|