1153 lines
56 KiB
Rust
1153 lines
56 KiB
Rust
//! Embedded MCP server. Claude Code (running inside the agent container)
|
|
//! connects to this over streamable-HTTP via `--mcp-config` (the long-lived
|
|
//! `hive-mcp-http` daemon); tool calls land here and are translated to
|
|
//! `Request::*` against hyperhive's own
|
|
//! per-container unix socket at `/run/hive/mcp.sock`.
|
|
//!
|
|
//! Two protocols, two surfaces:
|
|
//! - **hyperhive socket** at `/run/hive/mcp.sock` — JSON-line, our
|
|
//! broker-routed protocol. Unaffected by this module.
|
|
//! - **MCP over HTTP** owned by this module — what claude actually speaks.
|
|
//!
|
|
//! One `AgentServer { socket }` struct for all roles.
|
|
//! Tool access is gated upstream by `--allowedTools` (derived from the
|
|
//! agent's `ToolGroup` config); the server itself is a dumb dispatcher.
|
|
//! All tools go through the same `run_tool_envelope` helper.
|
|
|
|
use std::future::Future;
|
|
use std::path::PathBuf;
|
|
|
|
use anyhow::Result;
|
|
use rmcp::{ServerHandler, handler::server::wrapper::Parameters, tool, tool_handler, tool_router};
|
|
|
|
mod args;
|
|
mod render;
|
|
|
|
pub use args::{
|
|
AckTodosUntilArgs, AckUntilArgs, AgentGetLooseEndsArgs, AnswerArgs, AskArgs,
|
|
CancelLooseEndArgs, CancelScheduleArgs, CreateRepoArgs, EditScheduleArgs,
|
|
FireScheduleNowArgs, GetAgentMetaArgs, GetHostJournalArgs, GetLogsArgs, KillArgs, RecvArgs,
|
|
RemindArgs, RequestInitConfigArgs, RequestSchedulePromptArgs, RestartArgs, SendArgs,
|
|
SetStatusArgs, StartArgs, UpdateArgs, UpdateMetaInputsArgs,
|
|
};
|
|
pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv};
|
|
|
|
use render::{
|
|
dial_agent_socket, format_matrix_summary, local_questions, local_reminders, local_todos,
|
|
loose_end_kind_label, mark_local_todo_done, mark_local_todos_done_until,
|
|
matrix_unread_summary, parse_loose_end_kind, render_loose_ends, reply_err,
|
|
};
|
|
|
|
/// Write (or remove) the status file in the agent's own `state/` directory.
|
|
/// Called by `AgentServer::set_status` for both agent and manager flavors
|
|
/// before dispatching the wire `SetStatus` request (which only triggers a
|
|
/// dashboard rescan on the host side — file I/O moved here because the
|
|
/// harness runs as the agent user and has write access to `state/`, whereas
|
|
/// hive-c0re's `hive-core` user does not after the privsep migration).
|
|
///
|
|
/// Mirrors the validation in `hive-c0re::limits::check_status_text` so
|
|
/// the file is never written with text the server would later reject (which
|
|
/// would leave a stale invalid entry on disk).
|
|
fn write_status_file(text: &str) -> Result<(), String> {
|
|
// 200 chars mirrors STATUS_MAX_CHARS in hive-c0re/src/limits.rs.
|
|
// Keep in sync if that constant changes.
|
|
const STATUS_MAX_CHARS: usize = 200;
|
|
let trimmed = text.trim();
|
|
if !trimmed.is_empty() {
|
|
if trimmed.contains('\n') || trimmed.contains('\r') {
|
|
return Err(
|
|
"set_status text must be a single line — write multi-line context to \
|
|
a file under your state/ dir and reference that path from the chip instead"
|
|
.to_owned(),
|
|
);
|
|
}
|
|
let len = trimmed.chars().count();
|
|
if len > STATUS_MAX_CHARS {
|
|
return Err(format!(
|
|
"set_status text too long ({len} chars, max {STATUS_MAX_CHARS}); trim to a short summary"
|
|
));
|
|
}
|
|
}
|
|
let path = crate::paths::state_dir().join("hyperhive-status");
|
|
let result = if trimmed.is_empty() {
|
|
std::fs::remove_file(&path).or_else(|e| {
|
|
if e.kind() == std::io::ErrorKind::NotFound {
|
|
Ok(())
|
|
} else {
|
|
Err(e)
|
|
}
|
|
})
|
|
} else {
|
|
std::fs::write(&path, format!("{trimmed}\n"))
|
|
};
|
|
result.map_err(|e| format!("set_status write failed: {e}"))
|
|
}
|
|
|
|
/// Common envelope around every MCP tool handler: pre-log → run →
|
|
/// post-log. Tool results stay clean — the inbox-status hint lives in
|
|
/// the wake prompt + UI header, not appended here.
|
|
pub async fn run_tool_envelope<F>(tool: &'static str, args: String, body: F) -> String
|
|
where
|
|
F: Future<Output = String>,
|
|
{
|
|
tracing::info!(tool, %args, "tool: request");
|
|
let result = body.await;
|
|
tracing::info!(tool, result = %result, "tool: result");
|
|
result
|
|
}
|
|
|
|
/// Unified MCP tool surface for both sub-agent and manager roles.
|
|
///
|
|
/// Both sockets speak the same `Request` / `Response` wire, so a single
|
|
/// `dispatch` call covers both sockets — the only real difference is which
|
|
/// socket path is used and which tools the flavor enables.
|
|
#[derive(Debug, Clone)]
|
|
pub struct AgentServer {
|
|
socket: PathBuf,
|
|
}
|
|
|
|
impl AgentServer {
|
|
#[must_use]
|
|
pub fn new(socket: PathBuf) -> Self {
|
|
Self { socket }
|
|
}
|
|
|
|
/// Issue any `Request` through the retry-aware client. Returns the raw
|
|
/// `Response` plus the retry count so tool handlers can annotate their
|
|
/// result (see `annotate_retries`).
|
|
///
|
|
/// Both sockets speak the same `Request` type, so this single method
|
|
/// covers both.
|
|
async fn dispatch(
|
|
&self,
|
|
req: hive_core_agent_sock::Request,
|
|
) -> (Result<hive_core_agent_sock::Response, anyhow::Error>, u32) {
|
|
// Ride out a hive-c0re restart rather than surface it: the caller
|
|
// here is claude, and a tool error costs a whole retry turn's
|
|
// tokens where 60s of in-daemon patience costs nothing.
|
|
match hive_sock_client::request_retried::<_, hive_core_agent_sock::Response>(
|
|
&self.socket,
|
|
&req,
|
|
hive_sock_client::Retry::RideOutRestart,
|
|
)
|
|
.await
|
|
{
|
|
Ok((r, n)) => (Ok(r), n),
|
|
Err(e) => (Err(e), 0),
|
|
}
|
|
}
|
|
}
|
|
|
|
// IMPORTANT: when adding a new `#[tool]` fn to this impl, also add
|
|
// its name to the matching `ToolGroup::tools()` slice in hive-sh4re.
|
|
// Claude Code's permission gate refuses uninlisted MCP tools in
|
|
// non-interactive `--print` mode with "permissions not granted yet".
|
|
#[tool_router]
|
|
impl AgentServer {
|
|
#[tool(
|
|
description = "Send a message to another hyperhive agent (or to the operator). \
|
|
Use this to talk to peers or to surface output for the human at the dashboard."
|
|
)]
|
|
async fn send(&self, Parameters(args): Parameters<SendArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let to = args.to.clone();
|
|
// Check per-agent allow-list (hyperhive.allowedRecipients). When no
|
|
// policy file is present (e.g. manager containers) the check is a no-op.
|
|
if let Err(refusal) = crate::send_allow::check_send_allowed(&to) {
|
|
return run_tool_envelope("send", log, async move { refusal }).await;
|
|
}
|
|
run_tool_envelope("send", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Send {
|
|
to: args.to,
|
|
body: args.body,
|
|
in_reply_to: args.in_reply_to,
|
|
})
|
|
.await;
|
|
annotate_retries(format_ack(resp, "send", format!("sent to {to}")), retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Surface a structured question to either the operator OR a peer agent. \
|
|
Returns immediately with a question id — do NOT wait inline. When the recipient \
|
|
answers, a system message with event `question_answered { id, question, answer, \
|
|
answerer }` lands in your inbox; handle it on a future turn. \n\n\
|
|
Recipient: omit `to` (or set `to: \"operator\"`) for the human operator on the \
|
|
dashboard. Set `to: \"<agent-name>\"` to ask a peer agent — they receive a \
|
|
`question_asked { id, asker, question, options, multi }` event in their inbox \
|
|
and answer via `mcp__hyperhive__answer`. \n\n\
|
|
`options` is advisory: pass a short fixed-choice list when applicable, otherwise \
|
|
leave empty for free text. Set `multi: true` to let the answerer pick multiple \
|
|
options (checkboxes on the dashboard, hint to the agent otherwise) — answer comes \
|
|
back as a comma-separated string. Set `ttl_seconds` to auto-cancel a \
|
|
no-longer-relevant question — on expiry the answer is `[expired]` (with \
|
|
`answerer: \"ttl-watchdog\"`) and the same `question_answered` event fires."
|
|
)]
|
|
async fn ask(&self, Parameters(args): Parameters<AskArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("ask", log, async move {
|
|
let to = match args.to.map(|t| hive_types::Ident::parse(&t)).transpose() {
|
|
Ok(to) => to,
|
|
Err(reason) => return format!("invalid `to` agent name: {reason}"),
|
|
};
|
|
let question = args.question;
|
|
let target = to
|
|
.as_ref()
|
|
.map_or_else(|| "operator".to_owned(), std::string::ToString::to_string);
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Ask {
|
|
question: question.clone(),
|
|
options: args.options,
|
|
multi: args.multi,
|
|
ttl_seconds: args.ttl_seconds,
|
|
to,
|
|
})
|
|
.await;
|
|
let s = match resp {
|
|
Ok(hive_core_agent_sock::Response::QuestionQueued { id }) => {
|
|
// Best-effort local questions-mirror record — a dial
|
|
// failure just means `get_loose_ends` won't show this
|
|
// row locally; the actual question is already queued
|
|
// in c0re regardless.
|
|
let _ = dial_agent_socket(&hive_agent_sock::Request::RecordAskedQuestion {
|
|
id,
|
|
target,
|
|
question,
|
|
})
|
|
.await;
|
|
format!(
|
|
"question queued (id={id}); answer will arrive as a system \
|
|
`question_answered` event in your inbox"
|
|
)
|
|
}
|
|
other => reply_err(other, "ask"),
|
|
};
|
|
annotate_retries(s, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Answer a question that was routed to YOU via a `question_asked` system \
|
|
event in your inbox. Pass the `id` from that event and your `answer` string. The \
|
|
answer will surface in the asker's inbox as a `question_answered { id, question, \
|
|
answer, answerer: <your-name> }` event. \n\n\
|
|
Authorisation is strict — you can only answer questions where you are the declared \
|
|
target (i.e. the asker did `ask(to: \"<your-name>\", ...)`). Trying to answer an \
|
|
operator-targeted question or a question addressed to a different agent will fail."
|
|
)]
|
|
async fn answer(&self, Parameters(args): Parameters<AnswerArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let id = args.id;
|
|
run_tool_envelope("answer", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Answer {
|
|
id,
|
|
answer: args.answer,
|
|
})
|
|
.await;
|
|
// Clear the local `answering` mirror row whenever this agent is
|
|
// done owing a reply for `id`: either a genuine success, or
|
|
// c0re telling us the question already resolved without us
|
|
// (the asker cancelled/answered it first, surfacing as an
|
|
// "already answered"/"not found" rejection) — in both cases
|
|
// nothing is still owed, so the row would otherwise linger
|
|
// stale. Any other rejection (e.g. wrong answerer) means the
|
|
// question is still genuinely outstanding, so leave it be.
|
|
let should_clear = match &resp {
|
|
Ok(hive_core_agent_sock::Response::Ok) => true,
|
|
Ok(hive_core_agent_sock::Response::Err { message }) => {
|
|
message.contains("not found") || message.contains("already answered")
|
|
}
|
|
_ => false,
|
|
};
|
|
if should_clear {
|
|
let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await;
|
|
}
|
|
annotate_retries(
|
|
format_ack(resp, "answer", format!("answered question {id}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Pop messages from this agent's inbox. Returns one or more messages, or \
|
|
an empty marker if nothing is waiting. Always an immediate 'anything pending?' peek \
|
|
— never blocks. \n\n\
|
|
**Single-message default**: with no args (or `max: 1`) you get the next message — \
|
|
same behaviour the harness uses to drive a turn. \n\n\
|
|
**When idle, prefer ending the turn over polling here.** Ending the turn is the \
|
|
ONLY path that observes an in-container todo wake (bash-task completions, matrix \
|
|
unread, forge activity) — that signal reaches the harness loop between turns, never \
|
|
a live `recv` call, so looping on `recv` while idle means sitting blind until \
|
|
something else wakes you. Ending the turn also checkpoints your session, and costs \
|
|
no latency vs. polling for a real inbox message either way — the broker wakes the \
|
|
next turn just as fast. \n\n\
|
|
**Batch drain**: pass `max: N` (capped at 5) to drain up to N messages in one \
|
|
round-trip. Use this when the wake prompt told you the inbox has more queued, or \
|
|
any time you expect a burst — one tool call beats N consecutive single recvs. Empty \
|
|
result reported the same way regardless of `max`. \n\n\
|
|
After popping, the result appends a `(N more message(s) pending …)` line whenever the \
|
|
inbox still has queued messages — so you know whether to drain again (or `ack_until`) \
|
|
without a separate status check. No line means the inbox is empty."
|
|
)]
|
|
async fn recv(&self, Parameters(args): Parameters<RecvArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("recv", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Recv {
|
|
wait_seconds: None,
|
|
max: args.max,
|
|
})
|
|
.await;
|
|
annotate_retries(format_recv(resp), retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Bulk-mark inbox messages handled: every message with broker id \
|
|
<= `up_to` (ids show as `[msg #<id>]` in recv output and wake prompts) is \
|
|
acked in one call — pending and already-delivered rows alike. Use this to \
|
|
clear a backlog you've already triaged (e.g. a redelivered flood after a \
|
|
container restart, or a pile of stale notifications) instead of draining it \
|
|
one recv at a time: note the highest `[msg #N]` you've seen, then \
|
|
`ack_until(up_to: N)`. Acked messages never redeliver. Only affects YOUR \
|
|
inbox rows; messages newer than `up_to` stay queued. Returns how many rows \
|
|
were newly acked."
|
|
)]
|
|
async fn ack_until(&self, Parameters(args): Parameters<AckUntilArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("ack_until", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::AckUntil { up_to: args.up_to })
|
|
.await;
|
|
let rendered = match resp {
|
|
Ok(hive_core_agent_sock::Response::Acked { count }) => {
|
|
format!("acked {count} message(s) up to id {}", args.up_to)
|
|
}
|
|
other => reply_err(other, "ack_until"),
|
|
};
|
|
annotate_retries(rendered, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "List loose ends pending against this agent: unanswered questions \
|
|
where you are the asker (waiting on someone) or the target (someone's waiting on \
|
|
you), pending reminders you scheduled, plus — for the manager only — pending \
|
|
approvals you submitted that the operator hasn't acted on yet (agents with the \
|
|
`approvals` tool group also see their own pending approvals). Also lists active \
|
|
local tasks published by external MCP daemons (e.g. running bash tasks). Cheap sweep, no args. Useful \
|
|
at turn start to remember what you owe / what's owed to you without scrolling \
|
|
inbox history. Output is a short bulleted list with ids, ages in seconds, and \
|
|
the relevant context. Each `question` or `reminder` row can be cancelled by \
|
|
passing its id + kind to `cancel_loose_end`. Empty result is reported clearly. \
|
|
Todos are capped at 40 rendered rows (newest first) so a large backlog never makes \
|
|
this call fail — a trailer line reports how many more are pending and suggests an \
|
|
`ack_todos_until` value to bulk-clear the rest in one call instead of triaging \
|
|
hundreds of ids individually.\n\
|
|
Pass `agent: \"<name>\"` to inspect a specific peer agent's threads. Direct \
|
|
child agents are always accessible. For non-children, the `query_agent_state` \
|
|
capability is required — without it the request is rejected with an error."
|
|
)]
|
|
async fn get_loose_ends(&self, Parameters(args): Parameters<AgentGetLooseEndsArgs>) -> String {
|
|
run_tool_envelope("get_loose_ends", String::new(), async move {
|
|
let is_self_query = args.agent.is_none();
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::GetLooseEnds { agent: args.agent })
|
|
.await;
|
|
// Extract the vec so we can augment before rendering.
|
|
let mut loose_ends = match resp {
|
|
Ok(hive_core_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends,
|
|
other => return annotate_retries(reply_err(other, "get_loose_ends"), retries),
|
|
};
|
|
// Prepend matrix unread entry for self-queries only (can't
|
|
// reach another agent's matrix daemon from here).
|
|
if is_self_query && let Some(unread_rooms) = matrix_unread_summary().await {
|
|
let total = u32::try_from(unread_rooms.len()).unwrap_or(u32::MAX);
|
|
if total > 0 {
|
|
let summary = format_matrix_summary(&unread_rooms);
|
|
loose_ends.insert(
|
|
0,
|
|
hive_sh4re::LooseEnd::UnreadMatrix {
|
|
rooms: total,
|
|
summary,
|
|
},
|
|
);
|
|
}
|
|
}
|
|
// Merge the harness's local todos (loose-ends v2) for self-queries.
|
|
// The harness owns the todo store in-container; another agent's
|
|
// todos aren't reachable from here (same as matrix above).
|
|
if is_self_query && let Some(todos) = local_todos().await {
|
|
loose_ends.extend(todos);
|
|
}
|
|
// Merge local pending reminders — same self-query-only
|
|
// restriction: a manager asking for a child's loose-ends no longer
|
|
// sees the child's reminders, matching the todos precedent above).
|
|
if is_self_query && let Some(reminders) = local_reminders().await {
|
|
loose_ends.extend(reminders);
|
|
}
|
|
// Merge local mirrored questions — same self-query-only
|
|
// restriction as todos/reminders above. c0re no longer sources
|
|
// `Question` rows for `for_agent`/`hive_wide` (see
|
|
// `hive-c0re::loose_ends` doc), so this is the only place a
|
|
// self-query sees its own questions now; a manager query for a
|
|
// child still sees that child's approvals (unaffected) but no
|
|
// longer its questions, matching the reminders precedent.
|
|
if is_self_query && let Some(questions) = local_questions().await {
|
|
loose_ends.extend(questions);
|
|
}
|
|
annotate_retries(render_loose_ends(&loose_ends), retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Set a free-text status string visible on the operator dashboard. \
|
|
Call this at the START of every task to describe what you're working on (e.g. \
|
|
`\"processing matrix messages\"`, `\"fixing bitburner crash\"`, `\"idle\"`). Pass an empty \
|
|
string to clear. The status is shown on your dashboard card and persists across \
|
|
harness restarts."
|
|
)]
|
|
async fn set_status(&self, Parameters(args): Parameters<SetStatusArgs>) -> String {
|
|
run_tool_envelope("set_status", args.text.clone(), async move {
|
|
if let Err(e) = write_status_file(&args.text) {
|
|
return e;
|
|
}
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::SetStatus { text: args.text })
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "set_status", "status updated".to_owned()),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Fetch identity + status metadata for an agent. Returns canonical \
|
|
`name`, the current `hyperhive_rev` hive-c0re is running against, `running` \
|
|
(bool — whether the container is currently up; when false, `status_text` and \
|
|
`status_set_at` are stale pre-stop values and should not be treated as live), \
|
|
and the target's self-reported `status` text (set via `set_status`) plus how \
|
|
long ago it was set. Also returns the hive + swarm display names (`hive_name`, \
|
|
`swarm_name`) when the operator has configured `services.hyperhive.{hiveName, \
|
|
swarmName}`; both lines omitted when unset. Pass `name` to query a peer (e.g. \
|
|
check whether iris is idle before pinging them); omit `name` to get your own \
|
|
identity stamp — handy for state files / commit messages / cross-agent \
|
|
attribution that won't drift across renames or session-continue boundaries \
|
|
where the system-prompt label could be stale. Status reads `<none>` when the \
|
|
target has never called `set_status` or has cleared it. Also returns \
|
|
`matrix_accounts` — a list of matrix identities the agent can act as \
|
|
(`name`, `user_id?`, `homeserver`); omitted for agents with no matrix \
|
|
provisioning."
|
|
)]
|
|
async fn get_agent_meta(&self, Parameters(args): Parameters<GetAgentMetaArgs>) -> String {
|
|
let log = args.name.clone().unwrap_or_else(|| "<self>".to_owned());
|
|
run_tool_envelope("get_agent_meta", log, async move {
|
|
let name = match args.name.map(|n| hive_types::Ident::parse(&n)).transpose() {
|
|
Ok(name) => name,
|
|
Err(reason) => return format!("invalid agent name: {reason}"),
|
|
};
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::GetAgentMeta { name })
|
|
.await;
|
|
annotate_retries(format_agent_meta(resp), retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Cancel an open thread you own — a `question` you asked (the \
|
|
asker gets `[cancelled by <you>]` as the answer and unblocks) or a `reminder` \
|
|
you scheduled (hard-deleted before it fires). `kind` is `\"question\"` or \
|
|
`\"reminder\"`; `id` is the row id from the matching `get_loose_ends` entry \
|
|
or the `question_queued` reply you got when you submitted. Auth: you can only \
|
|
cancel rows where you're the asker / owner. Returns `ok` or an error string.\n\
|
|
`kind` may also be `\"approval\"` to withdraw a pending approval you submitted \
|
|
(before the operator acts on it) — root agent (`ruth`) only; the server rejects \
|
|
`approval` kind for all other callers.\n\
|
|
`kind` may also be `\"todo\"` to clear one of your own loose-ends-v2 todos \
|
|
(bash-task completions, matrix unread, forge activity — the id in the \
|
|
`get_loose_ends` `todo #N` line) — this dials the in-container socket directly, \
|
|
no bash task involved, so it's safe to call repeatedly without spawning more \
|
|
todos."
|
|
)]
|
|
async fn cancel_loose_end(&self, Parameters(args): Parameters<CancelLooseEndArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let id = args.id;
|
|
run_tool_envelope("cancel_loose_end", log, async move {
|
|
if args.kind.trim().eq_ignore_ascii_case("todo") {
|
|
return match mark_local_todo_done(id).await {
|
|
Some(hive_agent_sock::Response::Acked { count }) if count > 0 => {
|
|
format!("cleared todo {id}")
|
|
}
|
|
Some(hive_agent_sock::Response::Acked { .. }) => {
|
|
format!("no such todo {id} (already cleared, or never existed)")
|
|
}
|
|
Some(hive_agent_sock::Response::Err { message }) => {
|
|
format!("cancel_loose_end failed: {message}")
|
|
}
|
|
Some(other) => format!("cancel_loose_end unexpected response: {other:?}"),
|
|
None => "cancel_loose_end: local todo socket unavailable \
|
|
(HIVE_AGENT_SOCKET unset or harness unreachable)"
|
|
.to_owned(),
|
|
};
|
|
}
|
|
let kind = match parse_loose_end_kind(&args.kind) {
|
|
Ok(k) => k,
|
|
Err(e) => return e,
|
|
};
|
|
let kind_label = loose_end_kind_label(kind);
|
|
// Reminders are harness-local — dial the in-agent socket
|
|
// directly instead of the broker; every other kind
|
|
// (question/approval) still lives in c0re.
|
|
if kind == hive_sh4re::CancelLooseEndKind::Reminder {
|
|
return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id })
|
|
.await
|
|
{
|
|
Some(hive_agent_sock::Response::Acked { count }) if count > 0 => {
|
|
format!("cancelled {kind_label} {id}")
|
|
}
|
|
Some(hive_agent_sock::Response::Acked { .. }) => {
|
|
format!("cancel_loose_end failed: no pending {kind_label} {id}")
|
|
}
|
|
Some(hive_agent_sock::Response::Err { message }) => {
|
|
format!("cancel_loose_end failed: {message}")
|
|
}
|
|
Some(other) => format!("cancel_loose_end unexpected response: {other:?}"),
|
|
None => "cancel_loose_end failed: in-agent socket unavailable".to_owned(),
|
|
};
|
|
}
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id })
|
|
.await;
|
|
if resp.is_ok() && kind == hive_sh4re::CancelLooseEndKind::Question {
|
|
// Best-effort — cancel is ownership-gated to the asker on
|
|
// the c0re side, so a successful cancel here always means
|
|
// *this* agent's own `asked` mirror row for `id`. Known gap
|
|
// (documented on `Questions`): the target isn't notified,
|
|
// so their `answering` row lingers until they call
|
|
// `answer()` or it goes stale.
|
|
let _ = dial_agent_socket(&hive_agent_sock::Request::ClearQuestion { id }).await;
|
|
}
|
|
annotate_retries(
|
|
format_ack(
|
|
resp,
|
|
"cancel_loose_end",
|
|
format!("cancelled {kind_label} {id}"),
|
|
),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Bulk-clear local todos (loose-ends v2 — bash-task completions, matrix \
|
|
unread, forge activity): every pending todo with `id <= up_to` is acked in one call, \
|
|
same shape as `ack_until` for the message inbox. Use this when `get_loose_ends` \
|
|
reports more todos than it renders individually (its truncation summary suggests a \
|
|
value) or whenever a backlog has piled up past the point of clearing ids one at a \
|
|
time — note the highest id you've actually triaged, or take the suggested value \
|
|
verbatim to clear the whole shown-as-hidden tail. Only affects YOUR todos; ids above \
|
|
`up_to` stay pending. Returns how many were newly acked."
|
|
)]
|
|
async fn ack_todos_until(&self, Parameters(args): Parameters<AckTodosUntilArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("ack_todos_until", log, async move {
|
|
match mark_local_todos_done_until(args.up_to).await {
|
|
Some(hive_agent_sock::Response::Acked { count }) => {
|
|
format!("acked {count} todo(s) up to id {}", args.up_to)
|
|
}
|
|
Some(hive_agent_sock::Response::Err { message }) => {
|
|
format!("ack_todos_until failed: {message}")
|
|
}
|
|
Some(other) => format!("ack_todos_until unexpected response: {other:?}"),
|
|
None => "ack_todos_until: local todo socket unavailable \
|
|
(HIVE_AGENT_SOCKET unset or harness unreachable)"
|
|
.to_owned(),
|
|
}
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Create a git repo through hive-c0re. You CANNOT create repos with your \
|
|
own forge token (creation is disabled) — this is the only path. The repo is created in \
|
|
the c0re-owned `agents` org, you're added as a write collaborator (not owner), and the \
|
|
default branch gets branch protection so merges require an operator-team approval — you \
|
|
cannot merge your own PRs. `repo` is a single name segment (letters, digits, `-`, `_`, \
|
|
`.`). Returns the new repo's full name + clone URL; clone it over \
|
|
`$HIVE_FORGE_URL/agents/<repo>.git` and push/open PRs as normal."
|
|
)]
|
|
async fn create_repo(&self, Parameters(args): Parameters<CreateRepoArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("create_repo", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::CreateRepo { repo: args.repo })
|
|
.await;
|
|
let s = match resp {
|
|
Ok(hive_core_agent_sock::Response::RepoCreated {
|
|
full_name,
|
|
clone_url,
|
|
}) => format!("created repo {full_name} — clone: {clone_url}"),
|
|
other => reply_err(other, "create_repo"),
|
|
};
|
|
annotate_retries(s, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Schedule a reminder that lands in this agent's own inbox at a future \
|
|
time (sender will appear as `reminder`). Use for self-paced follow-ups: 'check task \
|
|
status in 60s', 'retry failed deploy at 14:00 UTC', 'nudge me when the operator's \
|
|
deploy window opens'. Set EXACTLY ONE of `delay_seconds` (fire N seconds from now) \
|
|
or `at_unix_timestamp` (fire at absolute epoch second). Body soft-caps at 4 KiB \
|
|
inline — anything larger gets auto-persisted to a file under your \
|
|
`/agents/<you>/state/reminders/` dir and the inbox message becomes a short pointer; \
|
|
pass `file_path` if you want to control the destination yourself. Returns \
|
|
immediately — the reminder lives in the broker until due."
|
|
)]
|
|
async fn remind(&self, Parameters(args): Parameters<RemindArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("remind", log, async move {
|
|
let timing = match (args.delay_seconds, args.at_unix_timestamp) {
|
|
(Some(_), Some(_)) => {
|
|
return "remind failed: pass exactly one of `delay_seconds` or \
|
|
`at_unix_timestamp`, not both"
|
|
.to_string();
|
|
}
|
|
(None, None) => {
|
|
return "remind failed: pass exactly one of `delay_seconds` or \
|
|
`at_unix_timestamp`"
|
|
.to_string();
|
|
}
|
|
(Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s },
|
|
(None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t },
|
|
};
|
|
// Reminders are harness-local — dial the in-agent socket
|
|
// directly instead of the broker.
|
|
match dial_agent_socket(&hive_agent_sock::Request::StoreReminder {
|
|
message: args.message,
|
|
timing,
|
|
file_path: args.file_path,
|
|
})
|
|
.await
|
|
{
|
|
Some(hive_agent_sock::Response::Ok) => "reminder scheduled".to_owned(),
|
|
Some(hive_agent_sock::Response::Err { message }) => {
|
|
format!("remind failed: {message}")
|
|
}
|
|
Some(other) => format!("remind unexpected response: {other:?}"),
|
|
None => "remind failed: in-agent socket unavailable".to_owned(),
|
|
}
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Compact the current session's context, mirroring the operator's \
|
|
dashboard `/compact` button. Gated: only honoured when this agent's last \
|
|
completed turn used more than 66% of the effective context window — below \
|
|
that the call is refused with an explanation and has no effect. On a pass, \
|
|
queues compaction for the end of the current turn (same deferred mechanism \
|
|
the dashboard button uses, so it never races a live claude process); the \
|
|
usual pre-compaction notes-checkpoint turn still fires first. No args."
|
|
)]
|
|
async fn compact(&self) -> String {
|
|
run_tool_envelope("compact", String::new(), async move {
|
|
match dial_agent_socket(&hive_agent_sock::Request::Compact).await {
|
|
Some(hive_agent_sock::Response::Ok) => {
|
|
"compact queued — will run at the end of the current turn".to_owned()
|
|
}
|
|
Some(hive_agent_sock::Response::Err { message }) => message,
|
|
Some(other) => format!("compact: unexpected response: {other:?}"),
|
|
None => "compact failed: in-agent socket unavailable".to_owned(),
|
|
}
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `lifecycle` tool group
|
|
// is granted to this agent. hive-c0re enforces the topology check
|
|
// server-side: the call is rejected unless `name` is a direct child.
|
|
#[tool(
|
|
description = "Restart a direct child sub-agent container (stop + start). \
|
|
Only succeeds if `name` is a direct child of this agent in the topology \
|
|
tree — the server enforces this. No approval required. \
|
|
Agents holding the `infra_admin` capability may also pass a hive \
|
|
infrastructure container name (`hive-ci`, `hive-gateway`, `hive-forge`) \
|
|
to restart it directly via the privileged helper."
|
|
)]
|
|
async fn restart(&self, Parameters(args): Parameters<RestartArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let name = args.name.clone();
|
|
run_tool_envelope("restart", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Restart { name: args.name })
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "restart", format!("restarted {name}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `lifecycle` tool group
|
|
// is granted to this agent. hive-c0re enforces the topology check
|
|
// server-side: the call is rejected unless `name` is a direct child.
|
|
#[tool(description = "Stop a direct child sub-agent container (graceful). \
|
|
Only succeeds if `name` is a direct child of this agent in the topology \
|
|
tree — the server enforces this. No approval required. \
|
|
State dir is kept; recreating the agent reuses prior config + credentials.")]
|
|
async fn kill(&self, Parameters(args): Parameters<KillArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let name = args.name.clone();
|
|
run_tool_envelope("kill", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Kill { name: args.name })
|
|
.await;
|
|
annotate_retries(format_ack(resp, "kill", format!("killed {name}")), retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `lifecycle` tool group
|
|
// is granted to this agent. hive-c0re enforces the topology check
|
|
// server-side: the call is rejected unless `name` is a direct child.
|
|
#[tool(
|
|
description = "Rebuild a direct child sub-agent: re-applies the current hyperhive \
|
|
flake + agent.nix and restarts the container. Only succeeds if `name` is a direct \
|
|
child of this agent in the topology tree — the server enforces this. \
|
|
No approval required. Idempotent — use when a child needs its config reapplied."
|
|
)]
|
|
async fn update(&self, Parameters(args): Parameters<UpdateArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let name = args.name.clone();
|
|
run_tool_envelope("update", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Update { name: args.name })
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "update", format!("updated {name}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `lifecycle` tool group
|
|
// is granted to this agent. Returns all topological descendants of the
|
|
// calling agent with their running status.
|
|
#[tool(
|
|
description = "List all containers that are topological descendants of this agent \
|
|
(direct children + their subtrees). Requires the `lifecycle` tool group. \
|
|
Returns every known descendant regardless of running state — check the `running` \
|
|
field to distinguish live from stopped containers. Ordered by topology depth \
|
|
(parents before children), then alphabetically within each tier."
|
|
)]
|
|
async fn list_containers(&self) -> String {
|
|
run_tool_envelope("list_containers", String::new(), async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::ListDescendants)
|
|
.await;
|
|
let body = match resp {
|
|
Ok(hive_core_agent_sock::Response::Containers { containers }) => {
|
|
if containers.is_empty() {
|
|
"no descendant containers".to_owned()
|
|
} else {
|
|
containers
|
|
.iter()
|
|
.map(|c| {
|
|
let status = if c.running { "running" } else { "stopped" };
|
|
format!("{} ({})", c.name, status)
|
|
})
|
|
.collect::<Vec<_>>()
|
|
.join("\n")
|
|
}
|
|
}
|
|
other => reply_err(other, "list_containers"),
|
|
};
|
|
annotate_retries(body, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is capability-gated (`read_host_journal`).
|
|
// It is added to `--allowedTools` by `allowed_capability_tools` only
|
|
// when `HIVE_CAPABILITIES` contains `read_host_journal`. hive-c0re
|
|
// performs a second capability check server-side before running journalctl.
|
|
#[tool(
|
|
description = "Fetch recent lines from the host journal (requires `read_host_journal` \
|
|
capability). All filters are optional - omit to get the last N host journal lines. \
|
|
`unit`: filter to a systemd unit (e.g. `hive-c0re.service`). \
|
|
`container`: nspawn machine name verbatim (e.g. `h-iris`). Omit for host journal. \
|
|
Agent containers use the `h-` prefix (e.g. `h-iris`, `h-atlas`); \
|
|
infrastructure containers use their full name (e.g. `hive-ci`, `hive-forge`, \
|
|
`hive-matrix`, `hive-gateway`). \
|
|
`lines`: how many lines (default 30, max 100). \
|
|
`priority`: minimum syslog level enum. \
|
|
`grep`: regex matched against log message fields (journalctl --grep). \
|
|
`since`: show entries on or newer than this (e.g. `-1h`, `2024-01-01 12:00:00`). \
|
|
`until`: show entries on or older than this."
|
|
)]
|
|
async fn get_host_journal(&self, Parameters(args): Parameters<GetHostJournalArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("get_host_journal", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::GetHostJournal {
|
|
unit: args.unit,
|
|
container: args.container,
|
|
lines: args.lines,
|
|
priority: args.priority,
|
|
grep: args.grep,
|
|
since: args.since,
|
|
until: args.until,
|
|
})
|
|
.await;
|
|
let result = match resp {
|
|
Ok(hive_core_agent_sock::Response::HostJournal { content }) => content,
|
|
other => reply_err(other, "get_host_journal"),
|
|
};
|
|
annotate_retries(result, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `approvals` tool group
|
|
// is configured for the agent (`HIVE_TOOL_GROUPS` contains `approvals`).
|
|
// hive-c0re performs a topology check server-side: only direct children
|
|
// of the calling agent are accepted; all other names are rejected.
|
|
#[tool(
|
|
description = "Initialise a brand-new direct child agent's proposed config repo and \
|
|
queue an `InitConfig` approval for the operator to review. Requires the `approvals` \
|
|
tool group. `name` must be a direct child of this agent in the topology tree. \
|
|
Fails if a config repo for that child already exists. On approval hive-c0re seeds \
|
|
`/agents/<name>/config/agent.nix` with the default template; customise + commit it, \
|
|
then the operator spawns the agent. Later config changes go through a PR on the \
|
|
child's `agent-configs/<name>` repo, reviewed + approved by the operator."
|
|
)]
|
|
async fn request_init_config(
|
|
&self,
|
|
Parameters(args): Parameters<RequestInitConfigArgs>,
|
|
) -> String {
|
|
let log = format!("{args:?}");
|
|
let name = args.name.clone();
|
|
run_tool_envelope("request_init_config", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::RequestInitConfig {
|
|
name: args.name,
|
|
description: args.description,
|
|
})
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(
|
|
resp,
|
|
"request_init_config",
|
|
format!("init_config approval queued for {name}"),
|
|
),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
// IMPORTANT: this tool is only available when the `lifecycle` tool group
|
|
// is granted to this agent. hive-c0re enforces the topology check
|
|
// server-side: the call is rejected unless `name` is a direct child.
|
|
#[tool(description = "Start a stopped direct child sub-agent container. \
|
|
Only succeeds if `name` is a direct child of this agent in the topology \
|
|
tree — the server enforces this. No approval required.")]
|
|
async fn start(&self, Parameters(args): Parameters<StartArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let name = args.name.clone();
|
|
run_tool_envelope("start", log, async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::Start { name: args.name })
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "start", format!("started {name}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Fetch recent journal log lines for a sub-agent container. Useful \
|
|
for diagnosing MCP server registration failures, startup crashes, plugin install \
|
|
errors, or any harness issue you can't see from inside the container. Pass the \
|
|
plain logical agent name (e.g. `gui`) — hive-c0re resolves the machine name. \
|
|
`lines` defaults to 50 (max capped at 500 on the host side)."
|
|
)]
|
|
async fn get_logs(&self, Parameters(args): Parameters<GetLogsArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
let agent = args.agent.clone();
|
|
run_tool_envelope("get_logs", log, async move {
|
|
let lines = args.lines.map(|n| n.min(500));
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::GetLogs {
|
|
agent: agent.clone(),
|
|
lines,
|
|
})
|
|
.await;
|
|
let s = match resp {
|
|
Ok(hive_core_agent_sock::Response::Logs { content }) => {
|
|
if content.is_empty() {
|
|
format!("(no journal output for {agent})")
|
|
} else {
|
|
content
|
|
}
|
|
}
|
|
other => reply_err(other, "get_logs"),
|
|
};
|
|
annotate_retries(s, retries)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Queue an approval for the operator to run `nix flake update` on the \
|
|
meta flake and commit the resulting lock changes. Pass specific input names to update \
|
|
only those inputs (e.g. `[\"bitburner-agent\"]`), or pass an empty list to update ALL \
|
|
inputs. Returns immediately — the lock update runs when the operator approves. \
|
|
Does NOT trigger container rebuilds — call `update` on each affected agent \
|
|
separately after the approval resolves."
|
|
)]
|
|
async fn request_update_meta_inputs(
|
|
&self,
|
|
Parameters(args): Parameters<UpdateMetaInputsArgs>,
|
|
) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("request_update_meta_inputs", log, async move {
|
|
let label = if args.inputs.is_empty() {
|
|
"all inputs".to_string()
|
|
} else {
|
|
args.inputs.join(", ")
|
|
};
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::RequestUpdateMetaInputs {
|
|
inputs: args.inputs,
|
|
description: args.description,
|
|
})
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(
|
|
resp,
|
|
"request_update_meta_inputs",
|
|
format!("approval queued: {label}"),
|
|
),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Queue an approval to add a scheduled prompt — one body delivered to \
|
|
N agent inboxes at a target time, optionally recurring every `interval_seconds`. \
|
|
The operator approves; on approve hive-c0re inserts the schedule and the worker \
|
|
fans it out. Even self-targeted schedules go through this flow (the operator pays \
|
|
for the wake-up tokens); the existing `remind` MCP tool stays the quick \
|
|
no-approval self-wake path. \n\n\
|
|
Catch-up clamp: if hive-c0re is down across multiple intervals, only ONE delayed \
|
|
fire happens on resume (per recurring schedule). The skipped-cycle count surfaces \
|
|
in the per-target `last_result` for the operator's audit trail. \n\n\
|
|
Per-target failure: a target name that doesn't resolve to a live agent → operator \
|
|
gets a one-line advisory `Message` from `system`; the schedule keeps firing for \
|
|
the other (live) targets."
|
|
)]
|
|
async fn request_schedule_prompt(
|
|
&self,
|
|
Parameters(args): Parameters<RequestSchedulePromptArgs>,
|
|
) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("request_schedule_prompt", log, async move {
|
|
let target_count = args.targets.len();
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::RequestSchedulePrompt(
|
|
hive_sh4re::SchedulePromptPayload {
|
|
targets: args.targets,
|
|
body: args.body,
|
|
first_fire_at_unix: args.first_fire_at_unix,
|
|
interval_seconds: args.interval_seconds,
|
|
description: args.description,
|
|
},
|
|
))
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(
|
|
resp,
|
|
"request_schedule_prompt",
|
|
format!("approval queued: {target_count} target(s)"),
|
|
),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Fire a scheduled prompt out of band — runs the per-target fan-out \
|
|
once immediately without disturbing the schedule's cadence. Recurring schedules \
|
|
keep their next_fire_at unchanged (the manual fire is additive). One-shot \
|
|
schedules are CONSUMED by the manual fire (cancelled afterwards): the operator's \
|
|
intent on a one-shot is 'send this now, the scheduled time was wrong'. \n\n\
|
|
Authorization mirrors `cancel_schedule`: you can fire your own schedules + any \
|
|
owned by a sub-agent in your subtree per topology.json."
|
|
)]
|
|
async fn fire_schedule_now(&self, Parameters(args): Parameters<FireScheduleNowArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("fire_schedule_now", log, async move {
|
|
let id = args.id;
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::FireScheduleNow { id })
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "fire_schedule_now", format!("fired #{id} now")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Cancel a scheduled prompt. With no `targets` field, cancels the \
|
|
whole schedule (all recipients flipped). With a non-empty `targets` list, cancels \
|
|
just those recipients; the schedule keeps firing for any remaining active targets \
|
|
and auto-cancels its parent row when every target is cancelled. \n\n\
|
|
Authorization: the manager can cancel its own schedules + any schedule whose \
|
|
owner is one of its sub-agents per topology.json. Other owners are refused."
|
|
)]
|
|
async fn cancel_schedule(&self, Parameters(args): Parameters<CancelScheduleArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("cancel_schedule", log, async move {
|
|
let id = args.id;
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::CancelSchedule {
|
|
id: args.id,
|
|
targets: args.targets,
|
|
})
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "cancel_schedule", format!("cancelled #{id}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "Edit an existing scheduled prompt's mutable fields. Pass only \
|
|
the fields you want to change — anything omitted keeps its current value. Editable: \
|
|
`body`, `description`, `interval_seconds` (positive only via this tool; flipping \
|
|
recurring→one-shot is operator-only via the dashboard), `next_fire_at_unix`, and \
|
|
the target set via `targets_add` / `targets_remove`. Both target lists are \
|
|
applied in the same transaction with removes-before-adds, so a single edit can \
|
|
swap a target atomically. Re-adding a previously-removed target starts a fresh \
|
|
per-target history (drops the tombstone). Draining all targets auto-cancels the \
|
|
parent schedule. \n\n\
|
|
Authorization mirrors `cancel_schedule` / `fire_schedule_now`: you can edit your \
|
|
own schedules + any owned by a sub-agent in your subtree per topology.json. \
|
|
Refuses cancelled schedules (the row's terminal — submit a fresh one)."
|
|
)]
|
|
async fn edit_schedule(&self, Parameters(args): Parameters<EditScheduleArgs>) -> String {
|
|
let log = format!("{args:?}");
|
|
run_tool_envelope("edit_schedule", log, async move {
|
|
let id = args.id;
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::EditSchedule {
|
|
id: args.id,
|
|
body: args.body,
|
|
description: args.description.map(Some),
|
|
interval_seconds: args.interval_seconds.map(Some),
|
|
next_fire_at_unix: args.next_fire_at_unix,
|
|
targets_add: args.targets_add,
|
|
targets_remove: args.targets_remove,
|
|
})
|
|
.await;
|
|
annotate_retries(
|
|
format_ack(resp, "edit_schedule", format!("edited #{id}")),
|
|
retries,
|
|
)
|
|
})
|
|
.await
|
|
}
|
|
|
|
#[tool(
|
|
description = "List every scheduled prompt in the queue (active + cancelled but \
|
|
not yet reaped). Returns the full snapshot — schedule id, owner, body, target set \
|
|
with per-target last_fired_at + last_result, next fire time, recurring interval. \
|
|
Use this to look up an id before calling `cancel_schedule`, or to audit what \
|
|
the swarm is going to be woken up about next."
|
|
)]
|
|
async fn list_schedules(&self) -> String {
|
|
run_tool_envelope("list_schedules", String::new(), async move {
|
|
let (resp, retries) = self
|
|
.dispatch(hive_core_agent_sock::Request::ListSchedules)
|
|
.await;
|
|
let body = match resp {
|
|
Ok(hive_core_agent_sock::Response::Schedules { schedules }) => {
|
|
serde_json::to_string(&schedules)
|
|
.unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}"))
|
|
}
|
|
other => reply_err(other, "list_schedules"),
|
|
};
|
|
annotate_retries(body, retries)
|
|
})
|
|
.await
|
|
}
|
|
}
|
|
|
|
#[tool_handler(
|
|
instructions = "You are a hyperhive agent. Use `send` to talk to peers (by their logical \
|
|
name) or to the operator (recipient `operator`). Use `recv` to drain your inbox one \
|
|
message at a time. Use `remind` to schedule a future wake-up message for yourself."
|
|
)]
|
|
impl ServerHandler for AgentServer {}
|
|
|
|
/// Run the MCP server over HTTP (rmcp streamable-http transport) on `addr`.
|
|
///
|
|
/// This is the sole transport for the built-in hyperhive surface. It runs as a
|
|
/// long-lived in-container daemon. claude reconnects
|
|
/// to the stable URL each turn instead of respawning and re-registering a stdio
|
|
/// subprocess, which removes the per-turn MCP registration race that can strand
|
|
/// an agent when the async `initialize`/`tools/list` loses to claude's first
|
|
/// tool call. `socket` is the hyperhive control socket every tool call dials
|
|
/// fresh (the handler holds only the path), so a host-side hive-c0re restart is
|
|
/// transparent — the next call just reconnects.
|
|
///
|
|
/// Binds loopback only in practice; the default `allowed_hosts`
|
|
/// (`localhost`/`127.0.0.1`/`::1`) rejects Host headers from anywhere else.
|
|
///
|
|
/// # Errors
|
|
///
|
|
/// Returns an error if the listener cannot bind `addr` or the HTTP server
|
|
/// exits with a fatal error.
|
|
pub async fn serve_http(socket: PathBuf, addr: std::net::SocketAddr) -> Result<()> {
|
|
use rmcp::transport::streamable_http_server::{
|
|
StreamableHttpServerConfig, StreamableHttpService, session::local::LocalSessionManager,
|
|
};
|
|
let session_manager = std::sync::Arc::new(LocalSessionManager::default());
|
|
let service = StreamableHttpService::new(
|
|
move || Ok(AgentServer::new(socket.clone())),
|
|
session_manager,
|
|
StreamableHttpServerConfig::default(),
|
|
);
|
|
let app = axum::Router::new().nest_service("/mcp", service);
|
|
let listener = tokio::net::TcpListener::bind(addr).await?;
|
|
tracing::info!(%addr, "serving hyperhive MCP over streamable-http at /mcp");
|
|
axum::serve(listener, app).await?;
|
|
Ok(())
|
|
}
|