From bf4937d1167f649c14c7670ad834fe622e04a380 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 1 Jun 2026 13:33:34 +0200 Subject: [PATCH] fix(#937): skip scheduled delivery if target already has pending scheduled message --- hive-c0re/src/broker.rs | 15 ++++++++++++++ hive-c0re/src/scheduled_prompts_worker.rs | 25 +++++++++++++++++++++++ 2 files changed, 40 insertions(+) diff --git a/hive-c0re/src/broker.rs b/hive-c0re/src/broker.rs index 0d16b1a6..020c07dd 100644 --- a/hive-c0re/src/broker.rs +++ b/hive-c0re/src/broker.rs @@ -263,6 +263,21 @@ impl Broker { Ok(u64::try_from(n.max(0)).unwrap_or(0)) } + /// Returns true when the recipient already has at least one + /// undelivered message from `sender` in the broker. Used by the + /// scheduler to avoid piling up repeated prompts when an agent is + /// slow or was briefly offline. + pub fn has_pending_from(&self, recipient: &str, sender: &str) -> Result { + let conn = self.conn.lock().unwrap(); + let n: i64 = conn.query_row( + "SELECT COUNT(*) FROM messages + WHERE recipient = ?1 AND sender = ?2 AND delivered_at IS NULL", + params![recipient, sender], + |row| row.get(0), + )?; + Ok(n > 0) + } + /// Long-poll variant of `recv_batch`: returns immediately if any /// row is pending (popping up to `max`); otherwise waits up to /// `timeout` for the broker to emit a `Sent { to: recipient }` diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs index 611a57d4..1f00e6a6 100644 --- a/hive-c0re/src/scheduled_prompts_worker.rs +++ b/hive-c0re/src/scheduled_prompts_worker.rs @@ -96,6 +96,31 @@ fn fire_schedule(coord: &Arc, schedule: &Schedule, now: i64) { notify_operator_missing_target(coord, schedule, target); continue; } + // Skip delivery if there is already an unread message from + // "scheduled" waiting in this target's inbox. Prevents prompt + // accumulation when the agent is slow to drain its queue or + // was briefly offline: at most one scheduled message is ever + // pending per target at a time. + match coord.broker.has_pending_from(target, "scheduled") { + Ok(true) => { + tracing::debug!( + schedule = schedule.id, + %target, + "scheduled_prompts: skipping — target already has pending scheduled message" + ); + let _ = coord.scheduled_prompts.record_target_result( + schedule.id, + target, + now, + "skipped: already pending", + ); + continue; + } + Err(e) => { + tracing::warn!(schedule = schedule.id, %target, error = ?e, "has_pending_from failed"); + } + Ok(false) => {} + } let msg = Message { from: "scheduled".to_owned(), to: target.clone(),