vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive-c0re stops owning scheduled prompts; agents lose the schedule tools

Browse source 
Scheduled prompts are swarm-level now (swarmctl schedule, #4936). The hive
keeps the rows it has only so the operator can copy them over.

- hive-c0re no longer fires anything: the scheduled-prompts worker is gone,
  and with it every create/edit/cancel/pause/resume/fire path (dashboard
  endpoints, agent-socket verbs, the SchedulePrompt approval kind and its
  approve handler). The store keeps open + list; GET /api/schedules stays.
- A v6 approvals migration drops schedule_prompt rows.
- Nothing submits approvals any more, so the submit path goes too:
  Approvals::submit_kind (now a test fixture), the ApprovalAdded dashboard
  event and its emitter.
- Coordinator::push_todo, whose only caller was the worker, goes, and with
  it hive_host_sock::agent_todo_socket and hive-c0re's hive-agent-sock /
  hive-sock-client deps.
- The schedules_changed dashboard event goes; nothing mutates the rows.
- Agents lose request_schedule_prompt, cancel_schedule, fire_schedule_now,
  edit_schedule and list_schedules, the `scheduling` tool group, their
  system-prompt line and their transcript formatters. `remind` is unchanged.
This commit is contained in:
atlas 2026-10-04 21:23:18 +02:00
commit c3a54bbdd0
31 changed files with 273 additions and 3124 deletions
Download patch file Download diff file Expand all files Collapse all files

6
hive-c0re/src/workers/mod.rs
View file

@ -1,6 +1,5 @@
//! Background tasks and periodic sweeps: crash/login watcher, the
//! scheduled-prompt delivery loop, boot-time auto-update
//! reconcile, the agent-sockets.json writer loop, the MCP socket listener
//! Background tasks and periodic sweeps: crash/login watcher, boot-time
//! auto-update reconcile, the agent-sockets.json writer loop, the MCP socket listener
//! reconcile loop, knowledge-repo sync, and the swarm wanted-state pull.
//! Each submodule is re-exported
//! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged.
@ -10,5 +9,4 @@ pub mod auto_update;
pub mod crash_watch;
pub mod knowledge;
pub mod mcp_sockets;
pub mod scheduled_prompts_worker;
pub mod wanted;

386
hive-c0re/src/workers/scheduled_prompts_worker.rs
View file

@ -1,386 +0,0 @@
//! Background loop that drains due `scheduled_prompts` rows and fans
//! the body to each active target. 5s poll cadence, shutdown-aware.
//! Catch-up clamp, missing-target handling, and broker-error retry
//! semantics: `docs/agent-lifecycle/approvals.md::Scheduled prompt worker`.
use std::sync::Arc;
use std::time::Duration;
use chrono::Utc;
use hive_sh4re::inbox::Message;
use crate::coordinator::Coordinator;
use crate::scheduled_prompts::Schedule;
/// Per-tick cap. Each schedule fires once per tick at most;
/// 100/tick × 5s tick = sustained throughput cap of ~20/sec,
/// matching `reminder_scheduler::REMINDER_BATCH_LIMIT`. Bump
/// together if real-world rates push past this.
const SCHEDULE_BATCH_LIMIT: u64 = 100;
/// Poll interval. Same 5s as the reminder scheduler — picking
/// up freshly-due rows within at most one tick keeps the
/// dashboard's "next fire in ..." countdown honest without
/// burning CPU on empty sweeps.
const POLL_INTERVAL: Duration = Duration::from_secs(5);
/// Reap cancelled schedules older than this from the table so
/// the dashboard list view doesn't accrue tombstones forever.
/// Cancelled rows live long enough that the operator can still
/// see what they cancelled in the recent past.
const CANCELLED_REAP_AGE: Duration = Duration::from_hours(1);
pub fn spawn(coord: Arc<Coordinator>) {
let mut shutdown = coord.shutdown_rx();
tokio::spawn(async move {
loop {
tick(&coord).await;
tokio::select! {
() = tokio::time::sleep(POLL_INTERVAL) => {}
_ = shutdown.changed() => {
tracing::info!("scheduled_prompts worker: shutdown signal received");
break;
}
}
}
});
}
async fn tick(coord: &Arc<Coordinator>) {
let now = Utc::now().timestamp();
let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
Ok(rows) => rows,
Err(e) => {
tracing::warn!(error = ?e, "scheduled_prompts: query due rows failed");
return;
}
};
if due.is_empty() {
// Periodic reaper still gets a chance even on empty ticks.
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
}
return;
}
for schedule in due {
fire_schedule(coord, &schedule, now).await;
}
let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
}
// Emit after all fires + reaps so the dashboard reflects updated
// last_fired_at_unix, next_fire_at_unix, and any reaped one-shots.
coord.emit_schedules_snapshot();
}
/// Fan out one schedule's body to every active target. Records
/// per-target `last_result`; advances or reaps the parent row at
/// the end depending on whether `interval_seconds` is set.
///
/// Delivery is `push_todo`, not a broker `Message` — a scheduled
/// prompt now wakes its target with a todo instead of driving an
/// immediate turn, by design: mara confirmed on the tracking issue
/// that this is the intended behavior, not an incidental side effect.
/// `key = "schedule:{id}"` per target gives `push_todo`'s own
/// upsert-by-key dedup the same job a now-removed
/// `has_pending_with_body` broker check used to do — collapsing a
/// re-fire of the *same schedule* against a target that hasn't
/// reviewed the last one yet, keyed on schedule identity rather than on
/// the body happening to be byte-identical. `deliver_to_target` also
/// passes `reopen_if_acked = true`, so this collapse only applies while
/// the previous fire is genuinely still sitting there un-reviewed — once
/// the target acks it, the *next* fire reopens the row and wakes again
/// even with an identical body, instead of silently staying quiet
/// forever — a recurring schedule with a static body used to wake its
/// target once, ever, and then look dead.
async fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
let known: std::collections::HashSet<String> = known_agents().await;
for target_row in &schedule.targets {
if target_row.cancelled_at_unix.is_some() {
continue;
}
let target = &target_row.target;
if !known.contains(target) {
let reason = format!("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);
continue;
}
let result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await {
Ok(()) => "ok".to_owned(),
Err(reason) => {
tracing::warn!(
schedule = schedule.id,
%target,
%reason,
"scheduled_prompts: delivery failed (will retry on next interval)"
);
reason
}
};
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");
}
}
// Advance or reap. One-shots delete; recurring re-arm with
// catch-up clamp.
if schedule.interval_seconds.is_some() {
match coord.scheduled_prompts.rearm(schedule.id, now) {
Ok(0) => {}
Ok(skipped) => {
tracing::info!(
schedule = schedule.id,
skipped,
"scheduled_prompts: caught up missed cycles"
);
}
Err(e) => {
tracing::warn!(error = ?e, schedule = schedule.id, "rearm failed");
}
}
} else if let Err(e) = coord.scheduled_prompts.delete(schedule.id) {
tracing::warn!(error = ?e, schedule = schedule.id, "delete one-shot failed");
}
}
/// Deliver `body` to a single already-known-live `target` via
/// `push_todo`, keyed on the schedule's own identity
/// (`schedule:{schedule_id}`) so a re-fire against a target that hasn't
/// reviewed the last one collapses via `push_todo`'s own upsert-by-key
/// dedup — but `reopen_if_acked = true` means that collapse only holds
/// while unreviewed; once acked, the next fire reopens regardless of
/// whether the body changed. Shared by the periodic `fire_schedule`
/// tick and the manual `fire_now` dashboard action — the only
/// difference between them is what each caller does with the `Result`
/// (log/prefix and per-target `last_result`/`FireNowReport`
/// bookkeeping), not the delivery choice itself.
async fn deliver_to_target(
coord: &Arc<Coordinator>,
schedule_id: i64,
target: &str,
body: &str,
) -> Result<(), String> {
coord
.push_todo(
target,
"schedule",
Some(format!("schedule:{schedule_id}")),
body.to_owned(),
Some("scheduled".to_owned()),
// Each fire is a distinct occurrence, not a restatement of
// a persisting condition — an agent that already acked the
// *previous* firing hasn't acked *this* one, so an acked
// row must reopen even when the body is byte-identical
// (the common case: most schedules don't vary their text
// per fire). See `Todos::upsert`'s doc comment for the
// full reconciler-vs-event rationale.
true,
)
.await
}
/// Send the operator a one-line advisory when a schedule fires
/// against an agent that no longer exists. Best-effort — failure
/// to send just gets logged; the schedule continues firing.
fn notify_operator_missing_target(coord: &Coordinator, schedule: &Schedule, target: &str) {
let body = format!(
"scheduled prompt #{id} fired but target `{target}` is not a live agent. \
body was:\n\n{body}",
id = schedule.id,
target = target,
body = schedule.body
);
let msg = Message {
from: hive_sh4re::manager::trusted_sender("scheduled"),
to: hive_sh4re::manager::OPERATOR_RECIPIENT.to_owned(),
body,
in_reply_to: None,
};
if let Err(e) = coord.broker.send(&msg) {
tracing::warn!(error = ?e, schedule = schedule.id, %target, "operator advisory send failed");
}
}
/// 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, utoipa::ToSchema)]
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,
/// Whether this manual fire re-armed a recurring schedule's timer
/// (`next_fire_at = now + interval`). `true` only when the caller
/// passed `reset_timer` AND the schedule is recurring.
pub timer_reset: bool,
}
/// Manual / out-of-band fire of a scheduled prompt ("fire now"
/// dashboard 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.
///
/// `reset_timer` re-arms a *recurring* schedule's countdown from now
/// (`next_fire_at = now + interval`) after the fan-out — the dashboard
/// fire-now dialog's "reset timer" checkbox. It's a no-op for one-shots
/// (still consumed) and when `false` (today's default: cadence intact).
///
/// 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<Coordinator>,
schedule_id: i64,
reset_timer: bool,
) -> anyhow::Result<FireNowReport> {
let now = Utc::now().timestamp();
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().await;
let mut report = FireNowReport {
ok: 0,
failed: 0,
missing: 0,
one_shot_consumed: false,
timer_reset: false,
};
for target_row in &schedule.targets {
if target_row.cancelled_at_unix.is_some() {
continue;
}
let target = &target_row.target;
if !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 result_str = match deliver_to_target(coord, schedule_id, target, &schedule.body).await {
Ok(()) => {
report.ok += 1;
"manual fire: ok".to_owned()
}
Err(reason) => {
report.failed += 1;
tracing::warn!(
schedule = schedule_id,
%target,
%reason,
"fire_now: delivery failed (no retry — manual fires don't loop)"
);
format!("manual fire: {reason}")
}
};
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");
}
}
match schedule.interval_seconds {
None => {
// One-shot is consumed by the manual fire. (reset_timer is
// moot here — there's no recurring cadence to re-arm.)
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;
}
}
Some(interval) if reset_timer => {
// Recurring + operator asked to reset: re-arm the countdown
// from now (now + interval), not along the existing cadence.
let next = now.saturating_add(i64::try_from(interval).unwrap_or(i64::MAX));
if let Err(e) = coord.scheduled_prompts.set_next_fire(schedule_id, next) {
tracing::warn!(error = ?e, schedule = schedule_id, "set_next_fire after manual fire reset failed");
} else {
report.timer_reset = true;
}
}
// Recurring without reset: cadence stays intact (additive fire).
Some(_) => {}
}
Ok(report)
}
/// Snapshot of live container names for the missing-target check.
/// Always seeds the manager name (which is always reachable);
/// adds every live nspawn container that matches the `h-` prefix.
/// On `lifecycle::list` failure the set stays at just the manager
/// — fail-CLOSED, meaning every non-manager target looks missing
/// this tick and gets the same treatment as a genuinely-destroyed
/// agent: operator advisory + per-target `last_result` annotation +
/// skipped delivery. Recurring
/// schedules recover automatically on the next tick (the lifecycle
/// listing usually works); one-shots that land on this window
/// lose their single delivery. Logged at `warn`, not propagated.
/// Shared by `fire_schedule` and `fire_now` — both are async now
/// (the sync `fire_schedule` used to need a `block_in_place` variant
/// of this before it started `push_todo`ing, which is itself async).
async fn known_agents() -> std::collections::HashSet<String> {
use std::collections::HashSet;
let mut out: HashSet<String> = HashSet::new();
out.insert(hive_sh4re::manager::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());
}
}
}
Err(e) => {
tracing::warn!(error = ?e, "fire_now: container listing failed");
}
}
out
}