From 3cd9240594755ac5badd3cb735624dd0f817ed5c Mon Sep 17 00:00:00 2001 From: damocles Date: Tue, 26 May 2026 13:43:04 +0200 Subject: [PATCH] scheduled prompts: fire-now operator + manager surfaces (closes #467) --- hive-ag3nt/prompts/manager.md | 1 + hive-ag3nt/src/mcp.rs | 35 ++++++ hive-c0re/src/dashboard.rs | 19 +++ hive-c0re/src/manager_server.rs | 45 ++++++- hive-c0re/src/scheduled_prompts_worker.rs | 144 ++++++++++++++++++++++ hive-sh4re/src/lib.rs | 8 ++ 6 files changed, 251 insertions(+), 1 deletion(-) diff --git a/hive-ag3nt/prompts/manager.md b/hive-ag3nt/prompts/manager.md index e79eda4e..f84490ca 100644 --- a/hive-ag3nt/prompts/manager.md +++ b/hive-ag3nt/prompts/manager.md @@ -13,6 +13,7 @@ Tools (hyperhive surface): - `mcp__hyperhive__request_update_meta_inputs(inputs?, description?)` — queue an approval for the operator to run `nix flake update [inputs...]` on the meta flake. Pass specific input names (e.g. `["bitburner-agent"]`) or omit / pass `[]` for all inputs. Returns immediately; lock update runs on operator approval. Does NOT trigger rebuilds — call `update(name)` on affected agents after approval resolves. - `mcp__hyperhive__request_schedule_prompt(targets, body, first_fire_at_unix, interval_seconds?, description?)` — queue an approval for the operator to add a scheduled prompt. On approve hive-c0re inserts a schedule row and the worker fans `body` out to each agent in `targets` at `first_fire_at_unix` (recurring every `interval_seconds` if set, one-shot when absent). Even self-targeted schedules go through approval — the existing `remind` tool stays the quick no-approval self-wake path. Catch-up clamp: long downtime fires ONCE per recurring row on resume (skipped count surfaces in per-target `last_result`), not N stacked pulses. - `mcp__hyperhive__cancel_schedule(id, targets?)` — cancel a schedule. Omit `targets` / pass empty to cancel the whole schedule; pass a list to cancel just those recipients (the schedule keeps firing for any remaining active targets, auto-cancels when every target is gone). Authorization: you can cancel schedules you own OR any owned by a sub-agent in your subtree per topology.json. +- `mcp__hyperhive__fire_schedule_now(id)` — fire a scheduled prompt out of band. Runs the per-target fan-out once immediately. Recurring schedules keep their cadence intact (the manual fire is additive); one-shot schedules are CONSUMED by the manual fire (cancelled afterwards). Same authorization as `cancel_schedule`. - `mcp__hyperhive__list_schedules()` — snapshot every schedule in the queue (active + cancelled-but-not-reaped). Returns id, owner, body, target set with per-target `last_fired_at` + `last_result`, `next_fire_at_unix`, recurring `interval_seconds`. Use to look up an id before cancelling, or to audit upcoming wake-ups across the swarm. - `mcp__hyperhive__get_logs(agent, lines?)` — fetch recent journal lines for a sub-agent container. Use to diagnose MCP-server registration failures, startup crashes, or harness issues you can't see from inside. Pass the plain logical agent name; `lines` defaults to 50 (capped at 500). - `mcp__hyperhive__ask(question, options?, multi?, ttl_seconds?, to?)` — surface a structured question to the operator (default, or `to: "operator"`) OR a sub-agent (`to: ""`). Returns immediately with a question id; the answer arrives later as a system `question_answered { id, question, answer, answerer }` event in your inbox. Options are advisory: the dashboard always lets the operator type a free-text answer in addition. Set `multi: true` to render options as checkboxes (operator can pick multiple); the answer comes back as `, `-separated. Set `ttl_seconds` to auto-cancel after a deadline (capped at 6h server-side) — on expiry the answer is `[expired]` and `answerer` is `"ttl-watchdog"`. Do not poll inside the same turn — finish the current work and react when the event lands. diff --git a/hive-ag3nt/src/mcp.rs b/hive-ag3nt/src/mcp.rs index 9d19013e..51e5b4b6 100644 --- a/hive-ag3nt/src/mcp.rs +++ b/hive-ag3nt/src/mcp.rs @@ -959,6 +959,14 @@ pub struct RequestSchedulePromptArgs { pub description: Option, } +#[derive(Debug, serde::Deserialize, schemars::JsonSchema)] +pub struct FireScheduleNowArgs { + /// Schedule id to fire out of band. Get this from a prior + /// `list_schedules` call or the approval-resolved event for + /// the originating `request_schedule_prompt`. + pub id: i64, +} + #[derive(Debug, serde::Deserialize, schemars::JsonSchema)] pub struct CancelScheduleArgs { /// Schedule id from a prior `list_schedules` call or the @@ -1269,6 +1277,33 @@ impl ManagerServer { .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, + ) -> String { + let log = format!("{args:?}"); + run_tool_envelope("fire_schedule_now", log, async move { + let id = args.id; + let (resp, retries) = self + .dispatch(hive_sh4re::ManagerRequest::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 \ diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index f4e322a6..800b30ae 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -76,6 +76,7 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { .route("/meta-update", post(post_meta_update)) .route("/api/schedules", get(api_schedules).post(post_schedule_new)) .route("/api/schedules/{id}/cancel", post(post_schedule_cancel)) + .route("/api/schedules/{id}/fire-now", post(post_schedule_fire_now)) .route("/dashboard/stream", get(dashboard_stream)) .route("/dashboard/history", get(dashboard_history)) // Anything not matched by the dynamic routes above falls @@ -1440,6 +1441,24 @@ async fn post_schedule_new( } } +/// `POST /api/schedules/{id}/fire-now` — operator-initiated +/// out-of-band fire of a scheduled prompt (#467). Runs the +/// per-target fan-out once immediately and reports per-target +/// outcome counts. Does NOT touch `next_fire_at_unix` on +/// recurring schedules (their cadence stays intact); one-shot +/// schedules are consumed (cancelled) by a manual fire — the +/// operator's intent is "send this now, the scheduled time was +/// wrong." +async fn post_schedule_fire_now( + State(state): State, + AxumPath(id): AxumPath, +) -> Response { + match crate::scheduled_prompts_worker::fire_now(&state.coord, id).await { + Ok(report) => axum::Json(report).into_response(), + Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")), + } +} + #[derive(serde::Deserialize, Default)] struct CancelScheduleForm { /// `None` / absent / empty array → cancel whole schedule. diff --git a/hive-c0re/src/manager_server.rs b/hive-c0re/src/manager_server.rs index 38331d50..9c4a5a52 100644 --- a/hive-c0re/src/manager_server.rs +++ b/hive-c0re/src/manager_server.rs @@ -358,6 +358,9 @@ async fn dispatch(req: &ManagerRequest, coord: &Arc) -> ManagerResp message: format!("list scheduled prompts: {e:#}"), }, }, + ManagerRequest::FireScheduleNow { id } => { + handle_fire_schedule_now(coord, hive_sh4re::MANAGER_AGENT, *id).await + } ManagerRequest::Ask { question, options, @@ -821,10 +824,50 @@ fn handle_cancel_schedule( } } +/// Authorize + dispatch a `FireScheduleNow` request from the +/// manager surface. Same ownership rules as `CancelSchedule`: +/// requester can fire its own schedules + any owned by an agent +/// in its subtree. The actual fan-out lives in +/// `scheduled_prompts_worker::fire_now`. +async fn handle_fire_schedule_now( + coord: &Arc, + requester: &str, + schedule_id: i64, +) -> ManagerResponse { + let schedule = match coord.scheduled_prompts.get(schedule_id) { + Ok(Some(s)) => s, + Ok(None) => { + return ManagerResponse::Err { + message: format!("schedule {schedule_id} not found"), + } + } + Err(e) => { + return ManagerResponse::Err { + message: format!("read schedule {schedule_id}: {e:#}"), + } + } + }; + if !cancel_authorized(requester, &schedule.owner) { + return ManagerResponse::Err { + message: format!( + "not authorized: {requester} cannot fire schedule owned by {owner}", + owner = schedule.owner + ), + }; + } + match crate::scheduled_prompts_worker::fire_now(coord, schedule_id).await { + Ok(_report) => ManagerResponse::Ok, + Err(e) => ManagerResponse::Err { + message: format!("fire schedule {schedule_id} now: {e:#}"), + }, + } +} + /// Permission check for `CancelSchedule` on the manager surface. /// `requester` (always `hm1nd` here) can cancel its own schedules. /// Sub-agent ownership is delegated to topology — see -/// `crate::topology::is_descendant_of`. +/// `crate::topology::is_descendant_of`. Also reused by +/// `handle_fire_schedule_now` — fire-auth follows the same shape. fn cancel_authorized(requester: &str, owner: &str) -> bool { if requester == owner { return true; diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs index 8880e9ea..2489274d 100644 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -245,3 +245,147 @@ fn now_unix() -> i64 { .and_then(|d| i64::try_from(d.as_secs()).ok()) .unwrap_or(0) } + +/// Per-target outcome counts for one `fire_now` invocation. +/// Returned to the operator so the dashboard can render +/// "fired to N (M failed, K missing)" without a follow-up GET. +#[derive(Debug, Clone, serde::Serialize)] +pub struct FireNowReport { + /// Targets the broker accepted the message for. + pub ok: u32, + /// Targets where broker.send returned an error. + pub failed: u32, + /// Targets that didn't resolve to a known agent (and got the + /// operator-advisory treatment). + pub missing: u32, + /// Whether the one-shot was consumed by this manual fire. + /// `true` only when the schedule was a one-shot (recurring + /// schedules never auto-cancel on manual fire — they keep + /// their cadence). + pub one_shot_consumed: bool, +} + +/// Manual / out-of-band fire of a scheduled prompt (#467 "fire +/// now" button). Mirrors the per-target fan-out of `fire_schedule` +/// but skips the rearm step entirely — manual fires don't disturb +/// a recurring schedule's rhythm. For one-shots, a manual fire +/// **consumes** the schedule (operator intent: "send this now, +/// the scheduled time was wrong"); recurring schedules keep their +/// `next_fire_at_unix` unchanged. +/// +/// `last_result` is annotated with the `manual fire:` prefix so +/// the dashboard's per-target last-result column can distinguish +/// scheduled fires from operator-initiated ones at a glance. +/// +/// Returns Err if the schedule is missing, cancelled, or fully +/// drained of active targets — the dashboard can surface those +/// as plain 4xxs instead of pretending to fire a phantom row. +pub async fn fire_now( + coord: &std::sync::Arc, + schedule_id: i64, +) -> anyhow::Result { + let now = now_unix(); + let schedule = coord + .scheduled_prompts + .get(schedule_id)? + .ok_or_else(|| anyhow::anyhow!("schedule {schedule_id} not found"))?; + if schedule.cancelled_at_unix.is_some() { + anyhow::bail!("schedule {schedule_id} is already cancelled"); + } + if !schedule.targets.iter().any(|t| t.cancelled_at_unix.is_none()) { + anyhow::bail!("schedule {schedule_id} has no active targets"); + } + let known = known_agents_async().await; + let mut report = FireNowReport { + ok: 0, + failed: 0, + missing: 0, + one_shot_consumed: false, + }; + for target_row in &schedule.targets { + if target_row.cancelled_at_unix.is_some() { + continue; + } + let target = &target_row.target; + if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) { + let reason = format!("manual fire: no such agent: {target}"); + if let Err(e) = coord.scheduled_prompts.record_target_result( + schedule_id, + target, + now, + &reason, + ) { + tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed"); + } + notify_operator_missing_target(coord, &schedule, target); + report.missing += 1; + continue; + } + let msg = Message { + from: "scheduled".to_owned(), + to: target.clone(), + body: schedule.body.clone(), + in_reply_to: None, + }; + let result = coord.broker.send(&msg); + let result_str = match &result { + Ok(()) => "manual fire: ok".to_owned(), + Err(e) => format!("manual fire: broker send failed: {e:#}"), + }; + if result.is_ok() { + report.ok += 1; + } else { + report.failed += 1; + tracing::warn!( + schedule = schedule_id, + %target, + error = ?result.as_ref().err(), + "fire_now: broker send failed (no retry — manual fires don't loop)" + ); + } + if let Err(e) = + coord + .scheduled_prompts + .record_target_result(schedule_id, target, now, &result_str) + { + tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed"); + } + } + if schedule.interval_seconds.is_none() { + // One-shot is consumed by the manual fire. Recurring + // schedules stay untouched — their cadence is the whole + // point and a manual fire is meant to be additive. + if let Err(e) = coord.scheduled_prompts.cancel_all(schedule_id) { + tracing::warn!(error = ?e, schedule = schedule_id, "cancel_all after one-shot manual fire failed"); + } else { + report.one_shot_consumed = true; + } + } + Ok(report) +} + +/// Async variant of `known_agents` for `fire_now`. Same logic + +/// same fail-closed degradation; the difference is just that the +/// dashboard handler is genuinely async so we `await` the +/// `lifecycle::list` directly instead of going through the +/// `block_in_place` shim. +async fn known_agents_async() -> std::collections::HashSet { + use std::collections::HashSet; + let mut out: HashSet = HashSet::new(); + out.insert(hive_sh4re::MANAGER_AGENT.to_owned()); + match crate::lifecycle::list().await { + Ok(list) => { + for raw in list { + if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) { + out.insert(name.to_owned()); + } else if raw == crate::lifecycle::MANAGER_NAME { + out.insert(hive_sh4re::MANAGER_AGENT.to_owned()); + } + } + } + Err(e) => { + tracing::warn!(error = ?e, "fire_now: container listing failed"); + } + } + out +} diff --git a/hive-sh4re/src/lib.rs b/hive-sh4re/src/lib.rs index b29003ac..e3dd9a40 100644 --- a/hive-sh4re/src/lib.rs +++ b/hive-sh4re/src/lib.rs @@ -940,6 +940,14 @@ pub enum ManagerRequest { /// unfiltered — the dashboard does the topology-filter for the /// per-agent view. ListSchedules, + /// Fire a scheduled prompt out of band (#467). Runs the + /// per-target fan-out once immediately without touching + /// `next_fire_at_unix` on recurring schedules; one-shots are + /// consumed by the manual fire. Authorization mirrors + /// `CancelSchedule`: the manager can fire its own schedules + /// + any owned by a sub-agent in its subtree per topology.json; + /// the operator surface bypasses the check. + FireScheduleNow { id: i64 }, } /// Submission payload for `RequestSchedulePrompt`. Lives outside the