hyperhive/hive-c0re/src/scheduled_prompts_worker.rs

//! Background loop that drains due `scheduled_prompts` rows and fans
//! the body to each active target. 5s poll cadence, shutdown-aware.
//! Catch-up clamp, missing-target handling, and broker-error retry
//! semantics: `docs/approvals.md::Scheduled prompt worker`.

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_hours(1);

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;
        }
        // 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: "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.
/// 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());
                }
                // Note: the old `else if raw == MANAGER_NAME` branch was dead code;
                // the manager container uses the h- prefix like all other agents.
            }
        }
        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)
}

/// Per-target outcome counts for one `fire_now` invocation.
/// Returned to the operator so the dashboard can render
/// "fired to N (M failed, K missing)" without a follow-up GET.
#[derive(Debug, Clone, serde::Serialize)]
pub struct FireNowReport {
    /// Targets the broker accepted the message for.
    pub ok: u32,
    /// Targets where broker.send returned an error.
    pub failed: u32,
    /// Targets that didn't resolve to a known agent (and got the
    /// operator-advisory treatment).
    pub missing: u32,
    /// Whether the one-shot was consumed by this manual fire.
    /// `true` only when the schedule was a one-shot (recurring
    /// schedules never auto-cancel on manual fire — they keep
    /// their cadence).
    pub one_shot_consumed: bool,
}

/// Manual / out-of-band fire of a scheduled prompt ("fire now"
/// dashboard button). Mirrors the per-target fan-out of `fire_schedule`
/// but skips the rearm step entirely — manual fires don't disturb
/// a recurring schedule's rhythm. For one-shots, a manual fire
/// **consumes** the schedule (operator intent: "send this now,
/// the scheduled time was wrong"); recurring schedules keep their
/// `next_fire_at_unix` unchanged.
///
/// `last_result` is annotated with the `manual fire:` prefix so
/// the dashboard's per-target last-result column can distinguish
/// scheduled fires from operator-initiated ones at a glance.
///
/// Returns Err if the schedule is missing, cancelled, or fully
/// drained of active targets — the dashboard can surface those
/// as plain 4xxs instead of pretending to fire a phantom row.
pub async fn fire_now(
    coord: &std::sync::Arc<Coordinator>,
    schedule_id: i64,
) -> anyhow::Result<FireNowReport> {
    let now = now_unix();
    let schedule = coord
        .scheduled_prompts
        .get(schedule_id)?
        .ok_or_else(|| anyhow::anyhow!("schedule {schedule_id} not found"))?;
    if schedule.cancelled_at_unix.is_some() {
        anyhow::bail!("schedule {schedule_id} is already cancelled");
    }
    if !schedule
        .targets
        .iter()
        .any(|t| t.cancelled_at_unix.is_none())
    {
        anyhow::bail!("schedule {schedule_id} has no active targets");
    }
    let known = known_agents_async().await;
    let mut report = FireNowReport {
        ok: 0,
        failed: 0,
        missing: 0,
        one_shot_consumed: false,
    };
    for target_row in &schedule.targets {
        if target_row.cancelled_at_unix.is_some() {
            continue;
        }
        let target = &target_row.target;
        if target != hive_sh4re::OPERATOR_RECIPIENT && !known.contains(target) {
            let reason = format!("manual fire: 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);
            report.missing += 1;
            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(()) => "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
                .record_target_result(schedule_id, target, now, &result_str)
        {
            tracing::warn!(error = ?e, schedule = schedule_id, %target, "record_target_result failed");
        }
    }
    if schedule.interval_seconds.is_none() {
        // One-shot is consumed by the manual fire. Recurring
        // schedules stay untouched — their cadence is the whole
        // point and a manual fire is meant to be additive.
        if let Err(e) = coord.scheduled_prompts.cancel_all(schedule_id) {
            tracing::warn!(error = ?e, schedule = schedule_id, "cancel_all after one-shot manual fire failed");
        } else {
            report.one_shot_consumed = true;
        }
    }
    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> {
    use std::collections::HashSet;
    let mut out: HashSet<String> = HashSet::new();
    out.insert(hive_sh4re::MANAGER_AGENT.to_owned());
    match crate::lifecycle::list().await {
        Ok(list) => {
            for raw in list {
                if let Some(name) = raw.strip_prefix(crate::lifecycle::AGENT_PREFIX) {
                    out.insert(name.to_owned());
                }
                // Note: the old `else if raw == MANAGER_NAME` branch was dead code;
                // the manager container uses the h- prefix like all other agents.
            }
        }
        Err(e) => {
            tracing::warn!(error = ?e, "fire_now: container listing failed");
        }
    }
    out
}