c0re: schedule_prompt approval kind + worker + manager surface (#444 step 2)

hive-c0re/src/scheduled_prompts_worker.rs

@@ -0,0 +1,232 @@
+//! Background loop that drains due `scheduled_prompts` rows
+//! (#444) and fans the body out as inbox `Message`s to each
+//! active target. Mirrors `reminder_scheduler::spawn` shape:
+//! single `spawn(coord)` entry, 5s poll cadence, shutdown-aware.
+//!
+//! ## Catch-up semantics
+//!
+//! When hive-c0re comes back from being down, a recurring row
+//! whose `next_fire_at` is well in the past would otherwise fire
+//! N delayed pulses in a row. Instead we fire ONCE and let
+//! `ScheduledPrompts::rearm` bump `next_fire_at` to the next
+//! interval slot ≥ `now`, recording the skipped-cycle count in
+//! the per-target `last_result` so operators see how many
+//! firings were caught up rather than losing the signal.
+//!
+//! ## Missing-target failure
+//!
+//! When a target name doesn't resolve to a known agent (the
+//! container has been destroyed, the operator typo'd a name,
+//! etc.) the worker:
+//!  1. records `last_result = "no such agent: <name>"` against
+//!     the per-target row,
+//!  2. sends a single advisory `Message` from `system` to
+//!     `operator` describing the schedule + target + reason,
+//!  3. continues fanning out to the other (live) targets.
+//!
+//! Transient broker errors (sqlite lock contention, etc.) are
+//! logged and retried on the next tick.
+
+use std::sync::Arc;
+use std::time::Duration;
+
+use hive_sh4re::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_secs(3600);
+
+pub fn spawn(coord: Arc<Coordinator>) {
+    let mut shutdown = coord.shutdown_rx();
+    tokio::spawn(async move {
+        loop {
+            tick(&coord);
+            tokio::select! {
+                () = tokio::time::sleep(POLL_INTERVAL) => {}
+                _ = shutdown.changed() => {
+                    tracing::info!("scheduled_prompts worker: shutdown signal received");
+                    break;
+                }
+            }
+        }
+    });
+}
+
+fn tick(coord: &Arc<Coordinator>) {
+    let now = now_unix();
+    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);
+    }
+    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");
+    }
+}
+
+/// 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);
+    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.
+        if target != hive_sh4re::OPERATOR_RECIPIENT && !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 msg = Message {
+            from: "scheduled".to_owned(),
+            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
+                .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");
+    }
+}
+
+/// Snapshot of live container names for the missing-target check.
+/// Returns an empty set on lifecycle errors — we fail-open then
+/// (every target gets through) and the broker rejects unknown
+/// recipients downstream.
+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();
+    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());
+                } else if raw == crate::lifecycle::MANAGER_NAME {
+                    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
+                }
+            }
+        }
+        Err(e) => {
+            tracing::warn!(error = ?e, "scheduled_prompts: container listing failed");
+        }
+    }
+    out
+}
+
+/// 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: "scheduled".to_owned(),
+        to: hive_sh4re::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");
+    }
+}
+
+fn now_unix() -> i64 {
+    std::time::SystemTime::now()
+        .duration_since(std::time::UNIX_EPOCH)
+        .ok()
+        .and_then(|d| i64::try_from(d.as_secs()).ok())
+        .unwrap_or(0)
+}