docs/approvals.md

@@ -203,22 +203,6 @@ One-shot rows fire once (if past due, on the next worker pass) and are deleted b
 
 `targets` is its own table (`scheduled_prompt_targets`) so partial cancellation flips a single row and the dashboard can show last-fired / last-result per recipient. Cancelling every target reaps the parent row on the next worker pass.
 
-### Scheduled prompt delivery: todo, not a broker message
-
-An agent target's delivery is `push_todo` (`Coordinator::push_todo`,
-`docs/coordinator.md` covers the mechanism generally), not a broker
-`Message` — a scheduled prompt wakes its target with a todo instead of
-driving an immediate turn, by design. `key = "schedule:<id>"` per
-target gives `push_todo`'s own upsert-by-key dedup the job a
-now-removed `has_pending_with_body` broker check used to do: a re-fire
-of the *same schedule* against a target that hasn't reviewed the last
-one collapses into that one todo instead of stacking up.
-
-**`operator` is the one exception** — it's a valid schedule target but
-has no in-container todo inbox, so it keeps the original broker
-`Message` path (the dashboard mirrors `to == operator` into its own
-pane, same as before).
-
 ### Missing-target failure
 
 When a target name doesn't resolve to a known agent (container
@@ -227,9 +211,7 @@ destroyed, operator typo, etc.) the worker:
 1. Records `last_result = "no such agent: <name>"` on the
    per-target row.
 2. Sends a single advisory `Message` from `system` to `operator`
-   naming the schedule, target, and reason. This one stays a `Message`
+   naming the schedule, target, and reason.
-   regardless of target type — it's a to-operator advisory about a
-   broken schedule, not the schedule's own delivery.
 3. Continues fanning out to the other live targets.
 
 Transient broker errors (sqlite lock contention, etc.) get the same

hive-c0re/src/actions.rs

@@ -739,7 +739,7 @@ async fn finish_approval(
     match approval.kind {
         ApprovalKind::InitConfig => {
             if ok {
-                let _ = coord
+                coord
                     .push_todo_submitter(
                         approval.id,
                         "core",
@@ -763,7 +763,7 @@ async fn finish_approval(
                     note.as_deref().unwrap_or("unknown error")
                 )
             };
-            let _ = coord
+            coord
                 .push_todo_submitter(
                     approval.id,
                     "core",
@@ -784,7 +784,7 @@ async fn finish_approval(
                 approval.fetched_sha.as_deref(),
                 terminal_tag.as_deref(),
             );
-            let _ = coord
+            coord
                 .push_todo_submitter(
                     approval.id,
                     "core",
@@ -963,7 +963,7 @@ pub async fn destroy(coord: &Arc<Coordinator>, name: &str, purge: bool) -> Resul
         tracing::warn!(%name, error = ?e, "agent_power: remove on destroy failed");
     }
     drop(guard);
-    let _ = coord
+    coord
         .push_todo(
             hive_sh4re::MANAGER_AGENT,
             "core",

hive-c0re/src/coordinator.rs

@@ -1387,14 +1387,6 @@ impl Coordinator {
     /// delivery. An agent that's down doesn't need a todo about
     /// something it'll never see appear this way; whatever mechanism
     /// resurfaces its state on the next boot is unrelated to this path.
-    ///
-    /// Returns `Ok(())` on a successful push and `Err(reason)` — a
-    /// short human-readable string, already logged at `debug`/`warn`
-    /// by this function — on any of the silent-no-op cases below.
-    /// Most callers are pure fire-and-forget notices and ignore it
-    /// (`let _ = coord.push_todo(...).await;`); callers that track a
-    /// per-target delivery outcome (e.g. the scheduled-prompts worker's
-    /// `last_result` column) use it instead of assuming success.
     pub async fn push_todo(
         &self,
         agent: &str,
@@ -1402,15 +1394,15 @@ impl Coordinator {
         key: Option<String>,
         summary: String,
         source: Option<String>,
-    ) -> Result<(), String> {
+    ) {
         let Ok(ident) = hive_types::Ident::parse(agent) else {
             tracing::warn!(%agent, "push_todo: not a valid agent ident, skipping");
-            return Err(format!("'{agent}' is not a valid agent ident"));
+            return;
         };
         let path = hive_host_sock::agent_todo_socket(&ident);
         if !path.exists() {
             tracing::debug!(%agent, path = %path.display(), "push_todo: agent socket not present (offline?), skipping");
-            return Err(format!("agent '{agent}' socket not present (offline?)"));
+            return;
         }
         let req = hive_agent_sock::Request::UpsertTodo {
             subsystem: subsystem.to_owned(),
@@ -1427,13 +1419,11 @@ impl Coordinator {
         {
             Ok(hive_agent_sock::Response::Err { message }) => {
                 tracing::warn!(%agent, %message, "push_todo: agent rejected the todo");
-                Err(format!("agent '{agent}' rejected the todo: {message}"))
             }
             Err(e) => {
                 tracing::warn!(%agent, error = ?e, "push_todo: dial failed");
-                Err(format!("push_todo dial to '{agent}' failed: {e:#}"))
             }
-            Ok(_) => Ok(()),
+            Ok(_) => {}
         }
     }
 
@@ -1446,10 +1436,10 @@ impl Coordinator {
         key: Option<String>,
         summary: String,
         source: Option<String>,
-    ) -> Result<(), String> {
+    ) {
         let target = self.submitter_or_manager(approval_id);
         self.push_todo(&target, subsystem, key, summary, source)
-            .await
+            .await;
     }
 
     /// Push a `HelperEvent` into an arbitrary agent's inbox. Encoded

hive-c0re/src/job_queue/exec.rs

@@ -152,7 +152,7 @@ async fn run_emit_rebuilt(coord: &Arc<Coordinator>, agent: &str, dag_id: Option<
         .then(|| dag_id.and_then(|dag| coord.job_queue.first_error(dag)))
         .flatten();
     let summary = crate::coordinator::rebuilt_todo_summary(agent, ok, note.as_deref(), None, None);
-    let _ = coord
+    coord
         .push_todo(
             hive_sh4re::MANAGER_AGENT,
             "core",
@@ -429,7 +429,7 @@ async fn run_stop(coord: &Arc<Coordinator>, name: &str) -> Result<()> {
     // own kind now.
     crate::lifecycle::kill(name).await?;
     coord.unregister_agent(name);
-    let _ = coord
+    coord
         .push_todo(
             hive_sh4re::MANAGER_AGENT,
             "core",

hive-c0re/src/server.rs

@@ -297,7 +297,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
             // Bind the MCP listener now that the container is starting up.
             // The harness connects to this socket on its first turn.
             coord.register_agent(name)?;
-            let _ = coord
+            coord
                 .push_todo(
                     hive_sh4re::MANAGER_AGENT,
                     "core",
@@ -312,7 +312,7 @@ async fn handle_spawn(coord: &Arc<Coordinator>, name: &str) -> Result<HostRespon
         Err(e) => {
             // Spawn failed: register_agent was never called, so there is
             // nothing to unregister. Notify the manager and propagate.
-            let _ = coord
+            coord
                 .push_todo(
                     hive_sh4re::MANAGER_AGENT,
                     "core",

hive-c0re/src/socket_server/lifecycle_handlers.rs

@@ -129,7 +129,7 @@ pub(super) async fn handle_kill(coord: &Arc<Coordinator>, agent: &str, name: &st
     .await;
     match result {
         Ok(()) => {
-            let _ = coord
+            coord
                 .push_todo(
                     hive_sh4re::MANAGER_AGENT,
                     "core",

hive-c0re/src/stores/broker.rs

@@ -354,6 +354,25 @@ 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` with exactly `body` in the
+    /// broker. Used by the scheduler to skip re-delivery of the same
+    /// scheduled prompt without blocking distinct schedules whose
+    /// bodies differ.
+    pub fn has_pending_with_body(&self, recipient: &str, sender: &str, body: &str) -> Result<bool> {
+        let conn = self.conn.lock().unwrap();
+        // An `ack_until`-closed pending row is dead — it must not
+        // suppress a fresh scheduled delivery of the same body.
+        let n: i64 = conn.query_row(
+            "SELECT COUNT(*) FROM messages
+              WHERE recipient = ?1 AND sender = ?2 AND body = ?3
+                AND delivered_at IS NULL AND acked_at IS NULL",
+            params![recipient, sender, body],
+            |row| row.get(0),
+        )?;
+        Ok(n > 0)
+    }
+
     /// Send a "your parent changed from X to Y" notification to `child`,
     /// coalescing with any existing undelivered one so that multiple moves
     /// while the agent is offline collapse into a single message spanning

hive-c0re/src/workers/crash_watch.rs

@@ -132,7 +132,7 @@ async fn emit_login_transitions(
 ) {
     for agent in current.difference(prev) {
         tracing::info!(%agent, "agent logged in");
-        let _ = coord
+        coord
             .push_todo(
                 hive_sh4re::MANAGER_AGENT,
                 "core",
@@ -163,7 +163,7 @@ async fn emit_login_transitions(
         .collect();
     for agent in current_needs.difference(&prev_needs) {
         tracing::info!(%agent, "agent needs login");
-        let _ = coord
+        coord
             .push_todo(
                 hive_sh4re::MANAGER_AGENT,
                 "core",

hive-c0re/src/workers/scheduled_prompts_worker.rs

@@ -34,7 +34,7 @@ pub fn spawn(coord: Arc<Coordinator>) {
     let mut shutdown = coord.shutdown_rx();
     tokio::spawn(async move {
         loop {
-            tick(&coord).await;
+            tick(&coord);
             tokio::select! {
                 () = tokio::time::sleep(POLL_INTERVAL) => {}
                 _ = shutdown.changed() => {
@@ -46,7 +46,7 @@ pub fn spawn(coord: Arc<Coordinator>) {
     });
 }
 
-async fn tick(coord: &Arc<Coordinator>) {
+fn tick(coord: &Arc<Coordinator>) {
     let now = Utc::now().timestamp();
     let due = match coord.scheduled_prompts.due(now, SCHEDULE_BATCH_LIMIT) {
         Ok(rows) => rows,
@@ -64,7 +64,7 @@ async fn tick(coord: &Arc<Coordinator>) {
         return;
     }
     for schedule in due {
-        fire_schedule(coord, &schedule, now).await;
+        fire_schedule(coord, &schedule, now);
     }
     let cutoff = now - i64::try_from(CANCELLED_REAP_AGE.as_secs()).unwrap_or(0);
     if let Err(e) = coord.scheduled_prompts.reap_cancelled(cutoff) {
@@ -78,28 +78,16 @@ async fn tick(coord: &Arc<Coordinator>) {
 /// 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.
-///
+fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
-/// Delivery is `push_todo`, not a broker `Message` — a scheduled
+    let known: std::collections::HashSet<String> = known_agents(coord);
-/// 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 — and does it more precisely (keyed on
-/// schedule identity, not on the body happening to be byte-identical).
-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;
-        // `operator` is a valid recipient (mara c4) — the operator has
+        // `operator` is a valid recipient (mara c4) — operator
-        // no in-container todo inbox, so it keeps the regular broker
+        // delivery uses the regular broker path; the dashboard
-        // `Message` path; the dashboard mirrors `to == operator` into
+        // mirrors `to == operator` into its own pane.
-        // its own pane.
         if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
             let reason = format!("no such agent: {target}");
             if let Err(e) =
@@ -112,18 +100,53 @@ async fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64)
             notify_operator_missing_target(coord, schedule, target);
             continue;
         }
-        let result_str = match deliver_to_target(coord, schedule.id, target, &schedule.body).await {
+        // Skip delivery if there is already an unread message from
+        // "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 — same body already pending for target"
+                );
+                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_with_body failed");
+            }
+            Ok(false) => {}
+        }
+        let msg = Message {
+            from: hive_sh4re::trusted_sender("scheduled"),
+            to: target.clone(),
+            body: schedule.body.clone(),
+            in_reply_to: None,
+        };
+        let result = coord.broker.send(&msg);
+        let result_str = match &result {
             Ok(()) => "ok".to_owned(),
-            Err(reason) => {
+            Err(e) => format!("broker send failed: {e:#}"),
+        };
+        if let Err(e) = result {
             tracing::warn!(
                 schedule = schedule.id,
                 %target,
-                    %reason,
+                error = ?e,
-                    "scheduled_prompts: delivery failed (will retry on next interval)"
+                "scheduled_prompts: broker send failed (will retry on next interval)"
             );
-                reason
         }
-        };
         if let Err(e) =
             coord
                 .scheduled_prompts
@@ -153,44 +176,43 @@ async fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64)
     }
 }
 
-/// Deliver `body` to a single already-known-live `target` — a broker
+/// Snapshot of live container names for the missing-target check.
-/// `Message` when `target` is the operator (no in-container todo inbox
+/// Always seeds the manager name (which is always reachable);
-/// to push into there), a `push_todo` otherwise, keyed on the
+/// adds every live nspawn container that matches the `h-` prefix.
-/// schedule's own identity (`schedule:{schedule_id}`) so a re-fire
+/// On `lifecycle::list` failure the set stays at just the manager
-/// against a target that hasn't reviewed the last one collapses via
+/// — fail-CLOSED, meaning every non-operator/non-manager target
-/// `push_todo`'s own upsert-by-key dedup. Shared by the periodic
+/// looks missing this tick and gets the same treatment as a
-/// `fire_schedule` tick and the manual `fire_now` dashboard action —
+/// genuinely-destroyed agent: operator advisory + per-target
-/// the only difference between them is what each caller does with the
+/// `last_result` annotation + skipped delivery. Recurring
-/// `Result` (log/prefix and per-target `last_result`/`FireNowReport`
+/// schedules recover automatically on the next tick (the lifecycle
-/// bookkeeping), not the delivery choice itself.
+/// listing usually works); one-shots that land on this window
-async fn deliver_to_target(
+/// lose their single delivery. Logged at `warn`, not propagated.
-    coord: &Arc<Coordinator>,
+fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> {
-    schedule_id: i64,
+    // `lifecycle::list` is async; the worker tick is sync. Use the
-    target: &str,
+    // blocking variant via a small `tokio::runtime::Handle::block_on`
-    body: &str,
+    // wrapper. The worker runs in its own tokio task so this is
-) -> Result<(), String> {
+    // safe (we're not in a `current_thread` runtime).
-    if target == hive_sh4re::OPERATOR_RECIPIENT {
+    use std::collections::HashSet;
-        let msg = Message {
+    let mut out: HashSet<String> = HashSet::new();
-            from: hive_sh4re::trusted_sender("scheduled"),
+    // Manager is always a scheduled-prompt target (fail-safe: include
-            to: target.to_owned(),
+    // it even if `list()` fails so prompts to the manager never silently drop).
-            body: body.to_owned(),
+    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
-            in_reply_to: None,
+    let containers = tokio::task::block_in_place(|| {
-        };
+        tokio::runtime::Handle::current().block_on(crate::lifecycle::list())
-        coord
+    });
-            .broker
+    match containers {
-            .send(&msg)
+        Ok(list) => {
-            .map_err(|e| format!("broker send failed: {e:#}"))
+            for raw in list {
-    } else {
+                if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
-        coord
+                    out.insert(name.to_owned());
-            .push_todo(
-                target,
-                "schedule",
-                Some(format!("schedule:{schedule_id}")),
-                body.to_owned(),
-                Some("scheduled".to_owned()),
-            )
-            .await
                 }
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "scheduled_prompts: container listing failed");
+        }
+    }
+    out
 }
 
 /// Send the operator a one-line advisory when a schedule fires
@@ -278,7 +300,7 @@ pub async fn fire_now(
     {
         anyhow::bail!("schedule {schedule_id} has no active targets");
     }
-    let known = known_agents().await;
+    let known = known_agents_async().await;
     let mut report = FireNowReport {
         ok: 0,
         failed: 0,
@@ -304,22 +326,28 @@ pub async fn fire_now(
             report.missing += 1;
             continue;
         }
-        let result_str = match deliver_to_target(coord, schedule_id, target, &schedule.body).await {
+        let msg = Message {
-            Ok(()) => {
+            from: hive_sh4re::trusted_sender("scheduled"),
+            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;
-                "manual fire: ok".to_owned()
+        } else {
-            }
-            Err(reason) => {
             report.failed += 1;
             tracing::warn!(
                 schedule = schedule_id,
                 %target,
-                    %reason,
+                error = ?result.as_ref().err(),
-                    "fire_now: delivery failed (no retry — manual fires don't loop)"
+                "fire_now: broker send failed (no retry — manual fires don't loop)"
             );
-                format!("manual fire: {reason}")
         }
-        };
         if let Err(e) =
             coord
                 .scheduled_prompts
@@ -354,21 +382,12 @@ pub async fn fire_now(
     Ok(report)
 }
 
-/// Snapshot of live container names for the missing-target check.
+/// Async variant of `known_agents` for `fire_now`. Same logic +
-/// Always seeds the manager name (which is always reachable);
+/// same fail-closed degradation; the difference is just that the
-/// adds every live nspawn container that matches the `h-` prefix.
+/// dashboard handler is genuinely async so we `await` the
-/// On `lifecycle::list` failure the set stays at just the manager
+/// `lifecycle::list` directly instead of going through the
-/// — fail-CLOSED, meaning every non-operator/non-manager target
+/// `block_in_place` shim.
-/// looks missing this tick and gets the same treatment as a
+async fn known_agents_async() -> std::collections::HashSet<String> {
-/// 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_AGENT.to_owned());