ag3nt: manager MCP tools + prompt for scheduled prompts (#444)

This commit is contained in:
damocles 2026-05-26 01:31:39 +02:00
commit 2593896383
3 changed files with 138 additions and 1 deletions

View file

@ -179,7 +179,8 @@ async fn serve(
| ManagerResponse::LooseEnds { .. }
| ManagerResponse::PendingRemindersCount { .. }
| ManagerResponse::ReminderRollup { .. }
| ManagerResponse::AgentMeta { .. },
| ManagerResponse::AgentMeta { .. }
| ManagerResponse::Schedules { .. },
) => {
tracing::warn!("recv produced unexpected response kind");
}

View file

@ -46,6 +46,9 @@ pub enum SocketReply {
QuestionQueued(i64),
Recent(Vec<hive_sh4re::InboxRow>),
Logs(String),
/// `list_schedules` result — used by the manager surface only;
/// AgentResponse has no equivalent variant.
Schedules(Vec<hive_sh4re::WireSchedule>),
LooseEnds(Vec<hive_sh4re::LooseEnd>),
PendingRemindersCount(u64),
ReminderRollup(hive_sh4re::ReminderStats),
@ -102,6 +105,7 @@ impl From<hive_sh4re::ManagerResponse> for SocketReply {
hive_sh4re::ManagerResponse::QuestionQueued { id } => Self::QuestionQueued(id),
hive_sh4re::ManagerResponse::Recent { rows } => Self::Recent(rows),
hive_sh4re::ManagerResponse::Logs { content } => Self::Logs(content),
hive_sh4re::ManagerResponse::Schedules { schedules } => Self::Schedules(schedules),
hive_sh4re::ManagerResponse::LooseEnds { loose_ends } => Self::LooseEnds(loose_ends),
hive_sh4re::ManagerResponse::PendingRemindersCount { count } => {
Self::PendingRemindersCount(count)
@ -929,6 +933,46 @@ pub struct UpdateMetaInputsArgs {
pub description: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct RequestSchedulePromptArgs {
/// Recipient agents — one schedule fires to many inboxes at the
/// scheduled time. `operator` is a legitimate target (mara: "we
/// want to get rid of the manager special case so yes manager
/// can be recipient" — the operator slot follows the same rule).
pub targets: Vec<String>,
/// Message body delivered to each target's inbox at fire time.
/// Same size budget as `send` bodies.
pub body: String,
/// Absolute unix timestamp (seconds) for the FIRST fire. For
/// recurring schedules the worker re-arms in
/// `interval_seconds` steps from this point on.
pub first_fire_at_unix: i64,
/// `None` / absent = one-shot. `Some(n > 0)` = recurring every
/// `n` seconds. The worker clamps catch-up so a long downtime
/// fires ONCE on resume (skipped-cycle count surfaces in the
/// per-target last_result), not N delayed pulses in a row.
#[serde(default)]
pub interval_seconds: Option<u64>,
/// Optional description shown on the dashboard approval card +
/// preserved on the schedule row for later operator reference.
#[serde(default)]
pub description: Option<String>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct CancelScheduleArgs {
/// Schedule id from a prior `list_schedules` call or the
/// approval-resolved event for a `request_schedule_prompt`.
pub id: i64,
/// Optional target list. `None` / empty = cancel the entire
/// schedule. `Some(["alice", "bob"])` = cancel just those
/// recipients (the schedule keeps firing for any remaining
/// active targets, and auto-cancels its parent row when every
/// target is gone).
#[serde(default)]
pub targets: Option<Vec<String>>,
}
#[derive(Debug, serde::Deserialize, schemars::JsonSchema)]
pub struct GetLogsArgs {
/// Logical agent name to fetch logs for (e.g. `gui`, `hm1nd`).
@ -1181,6 +1225,95 @@ impl ManagerServer {
.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_sh4re::ManagerRequest::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 = "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_sh4re::ManagerRequest::CancelSchedule {
id: args.id,
targets: args.targets,
})
.await;
annotate_retries(format_ack(resp, "cancel_schedule", format!("cancelled #{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_sh4re::ManagerRequest::ListSchedules).await;
let body = match resp {
Ok(SocketReply::Schedules(schedules)) => serde_json::to_string(&schedules)
.unwrap_or_else(|e| format!("list_schedules: serialise: {e:#}")),
Ok(SocketReply::Err(m)) => format!("list_schedules: {m}"),
Ok(other) => format!("list_schedules unexpected response: {other:?}"),
Err(e) => format!("list_schedules transport error: {e:#}"),
};
annotate_retries(body, retries)
})
.await
}
#[tool(
description = "Surface a structured question to either the operator OR a sub-agent. \
Returns immediately with a question id do NOT wait inline. When the recipient \