From 3283362b361702d87209e66e25f4021688d61b42 Mon Sep 17 00:00:00 2001
From: damocles <damocles@hyperhive>
Date: Mon, 1 Jun 2026 14:43:12 +0200
Subject: [PATCH] =?UTF-8?q?fix(#937):=20use=20has=5Fpending=5Fwith=5Fbody?=
 =?UTF-8?q?=20in=20worker=20=E2=80=94=20mirrors=20broker=20change?=
MIME-Version: 1.0
Content-Type: text/plain; charset=UTF-8
Content-Transfer-Encoding: 8bit

---
 hive-c0re/src/scheduled_prompts_worker.rs | 15 ++++++++-------
 1 file changed, 8 insertions(+), 7 deletions(-)

diff --git a/hive-c0re/src/scheduled_prompts_worker.rs b/hive-c0re/src/scheduled_prompts_worker.rs
index 1f00e6a6..a37f363a 100644
--- a/hive-c0re/src/scheduled_prompts_worker.rs
+++ b/hive-c0re/src/scheduled_prompts_worker.rs
@@ -97,16 +97,16 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
             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") {
+        // "scheduled" waiting in this target's inbox. Prevents the same
+        // scheduled prompt from stacking up when an agent is slow or
+        // briefly offline, while still allowing distinct scheduled
+        // messages (different body) to enqueue independently.
+        match coord.broker.has_pending_with_body(target, "scheduled", &schedule.body) {
             Ok(true) => {
                 tracing::debug!(
                     schedule = schedule.id,
                     %target,
-                    "scheduled_prompts: skipping — target already has pending scheduled message"
+                    "scheduled_prompts: skipping — same body already pending for target"
                 );
                 let _ = coord.scheduled_prompts.record_target_result(
                     schedule.id,
@@ -117,7 +117,7 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
                 continue;
             }
             Err(e) => {
-                tracing::warn!(schedule = schedule.id, %target, error = ?e, "has_pending_from failed");
+                tracing::warn!(schedule = schedule.id, %target, error = ?e, "has_pending_with_body failed");
             }
             Ok(false) => {}
         }
@@ -384,3 +384,4 @@ async fn known_agents_async() -> std::collections::HashSet<String> {
     }
     out
 }
+