hive-c0re: deliver scheduled prompts via push_todo, not a broker message

docs/approvals.md

@@ -203,6 +203,22 @@ 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
@@ -211,7 +227,9 @@ 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.
+   naming the schedule, target, and reason. This one stays a `Message`
+   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 {
-                coord
+                let _ = coord
                     .push_todo_submitter(
                         approval.id,
                         "core",
@@ -763,7 +763,7 @@ async fn finish_approval(
                     note.as_deref().unwrap_or("unknown error")
                 )
             };
-            coord
+            let _ = coord
                 .push_todo_submitter(
                     approval.id,
                     "core",
@@ -784,7 +784,7 @@ async fn finish_approval(
                 approval.fetched_sha.as_deref(),
                 terminal_tag.as_deref(),
             );
-            coord
+            let _ = 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);
-    coord
+    let _ = coord
         .push_todo(
             hive_sh4re::MANAGER_AGENT,
             "core",

hive-c0re/src/coordinator.rs

@@ -1387,6 +1387,14 @@ 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,
@@ -1394,15 +1402,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;
+            return Err(format!("'{agent}' is not a valid agent ident"));
         };
         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;
+            return Err(format!("agent '{agent}' socket not present (offline?)"));
         }
         let req = hive_agent_sock::Request::UpsertTodo {
             subsystem: subsystem.to_owned(),
@@ -1419,11 +1427,13 @@ 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(()),
         }
     }
 
@@ -1436,10 +1446,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);
-    coord
+    let _ = 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);
-    coord
+    let _ = 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)?;
-            coord
+            let _ = 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.
-            coord
+            let _ = 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(()) => {
-            coord
+            let _ = coord
                 .push_todo(
                     hive_sh4re::MANAGER_AGENT,
                     "core",

hive-c0re/src/stores/broker.rs

@@ -354,25 +354,6 @@ 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");
-        coord
+        let _ = 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");
-        coord
+        let _ = 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);
+            tick(&coord).await;
             tokio::select! {
                 () = tokio::time::sleep(POLL_INTERVAL) => {}
                 _ = shutdown.changed() => {
@@ -46,7 +46,7 @@ pub fn spawn(coord: Arc<Coordinator>) {
     });
 }
 
-fn tick(coord: &Arc<Coordinator>) {
+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,
@@ -64,7 +64,7 @@ fn tick(coord: &Arc<Coordinator>) {
         return;
     }
     for schedule in due {
-        fire_schedule(coord, &schedule, now);
+        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) {
@@ -78,16 +78,28 @@ 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) {
-    let known: std::collections::HashSet<String> = known_agents(coord);
+///
+/// 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 — 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) — operator
-        // delivery uses the regular broker path; the dashboard
-        // mirrors `to == operator` into its own pane.
+        // `operator` is a valid recipient (mara c4) — the operator has
+        // no in-container todo inbox, so it keeps the regular broker
+        // `Message` path; the dashboard mirrors `to == operator` into
+        // its own pane.
         if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
             let reason = format!("no such agent: {target}");
             if let Err(e) =
@@ -100,53 +112,18 @@ fn fire_schedule(coord: &Arc<Coordinator>, 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 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!(
+        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,
-                    "scheduled_prompts: skipping — same body already pending for target"
+                    %reason,
+                    "scheduled_prompts: delivery failed (will retry on next interval)"
                 );
-                let _ = coord.scheduled_prompts.record_target_result(
-                    schedule.id,
-                    target,
-                    now,
-                    "skipped: already pending",
-                );
-                continue;
+                reason
             }
-            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(e) => format!("broker send failed: {e:#}"),
-        };
-        if let Err(e) = result {
-            tracing::warn!(
-                schedule = schedule.id,
-                %target,
-                error = ?e,
-                "scheduled_prompts: broker send failed (will retry on next interval)"
-            );
-        }
         if let Err(e) =
             coord
                 .scheduled_prompts
@@ -176,43 +153,44 @@ fn fire_schedule(coord: &Arc<Coordinator>, schedule: &Schedule, now: i64) {
     }
 }
 
-/// 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-operator/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.
-fn known_agents(_coord: &Coordinator) -> std::collections::HashSet<String> {
-    // `lifecycle::list` is async; the worker tick is sync. Use the
-    // blocking variant via a small `tokio::runtime::Handle::block_on`
-    // wrapper. The worker runs in its own tokio task so this is
-    // safe (we're not in a `current_thread` runtime).
-    use std::collections::HashSet;
-    let mut out: HashSet<String> = HashSet::new();
-    // Manager is always a scheduled-prompt target (fail-safe: include
-    // it even if `list()` fails so prompts to the manager never silently drop).
-    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
-    let containers = tokio::task::block_in_place(|| {
-        tokio::runtime::Handle::current().block_on(crate::lifecycle::list())
-    });
-    match containers {
-        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, "scheduled_prompts: container listing failed");
-        }
+/// Deliver `body` to a single already-known-live `target` — a broker
+/// `Message` when `target` is the operator (no in-container todo inbox
+/// to push into there), a `push_todo` otherwise, 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. 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> {
+    if target == hive_sh4re::OPERATOR_RECIPIENT {
+        let msg = Message {
+            from: hive_sh4re::trusted_sender("scheduled"),
+            to: target.to_owned(),
+            body: body.to_owned(),
+            in_reply_to: None,
+        };
+        coord
+            .broker
+            .send(&msg)
+            .map_err(|e| format!("broker send failed: {e:#}"))
+    } else {
+        coord
+            .push_todo(
+                target,
+                "schedule",
+                Some(format!("schedule:{schedule_id}")),
+                body.to_owned(),
+                Some("scheduled".to_owned()),
+            )
+            .await
     }
-    out
 }
 
 /// Send the operator a one-line advisory when a schedule fires
@@ -300,7 +278,7 @@ pub async fn fire_now(
     {
         anyhow::bail!("schedule {schedule_id} has no active targets");
     }
-    let known = known_agents_async().await;
+    let known = known_agents().await;
     let mut report = FireNowReport {
         ok: 0,
         failed: 0,
@@ -326,28 +304,22 @@ pub async fn fire_now(
             report.missing += 1;
             continue;
         }
-        let msg = Message {
-            from: hive_sh4re::trusted_sender("scheduled"),
-            to: target.clone(),
-            body: schedule.body.clone(),
-            in_reply_to: None,
+        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}")
+            }
         };
-        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
@@ -382,12 +354,21 @@ pub async fn fire_now(
     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<String> {
+/// 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-operator/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_AGENT.to_owned());