hyperhive/hive-c0re/src/workers/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/agent-lifecycle/approvals.md::Scheduled prompt worker`.

use std::sync::Arc;
use std::time::Duration;

use chrono::Utc;
use hive_sh4re::inbox::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).await;
            tokio::select! {
                () = tokio::time::sleep(POLL_INTERVAL) => {}
                _ = shutdown.changed() => {
                    tracing::info!("scheduled_prompts worker: shutdown signal received");
                    break;
                }
            }
        }
    });
}

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,
        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).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) {
        tracing::warn!(error = ?e, "scheduled_prompts: reap cancelled failed");
    }
    // Emit after all fires + reaps so the dashboard reflects updated
    // last_fired_at_unix, next_fire_at_unix, and any reaped one-shots.
    coord.emit_schedules_snapshot();
}

/// 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.
///
/// 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, keyed on schedule identity rather than on
/// the body happening to be byte-identical. `deliver_to_target` also
/// passes `reopen_if_acked = true`, so this collapse only applies while
/// the previous fire is genuinely still sitting there un-reviewed — once
/// the target acks it, the *next* fire reopens the row and wakes again
/// even with an identical body, instead of silently staying quiet
/// forever — a recurring schedule with a static body used to wake its
/// target once, ever, and then look dead.
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
        // 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::manager::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 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,
                    %reason,
                    "scheduled_prompts: delivery failed (will retry on next interval)"
                );
                reason
            }
        };
        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");
    }
}

/// 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 — but `reopen_if_acked = true`
/// means that collapse only holds while unreviewed; once acked, the next
/// fire reopens regardless of whether the body changed. 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::manager::OPERATOR_RECIPIENT {
        let msg = Message {
            from: hive_sh4re::manager::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()),
                // Each fire is a distinct occurrence, not a restatement of
                // a persisting condition — an agent that already acked the
                // *previous* firing hasn't acked *this* one, so an acked
                // row must reopen even when the body is byte-identical
                // (the common case: most schedules don't vary their text
                // per fire). See `Todos::upsert`'s doc comment for the
                // full reconciler-vs-event rationale.
                true,
            )
            .await
    }
}

/// 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: hive_sh4re::manager::trusted_sender("scheduled"),
        to: hive_sh4re::manager::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");
    }
}

/// 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, utoipa::ToSchema)]
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,
    /// Whether this manual fire re-armed a recurring schedule's timer
    /// (`next_fire_at = now + interval`). `true` only when the caller
    /// passed `reset_timer` AND the schedule is recurring.
    pub timer_reset: 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.
///
/// `reset_timer` re-arms a *recurring* schedule's countdown from now
/// (`next_fire_at = now + interval`) after the fan-out — the dashboard
/// fire-now dialog's "reset timer" checkbox. It's a no-op for one-shots
/// (still consumed) and when `false` (today's default: cadence intact).
///
/// 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,
    reset_timer: bool,
) -> anyhow::Result<FireNowReport> {
    let now = Utc::now().timestamp();
    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().await;
    let mut report = FireNowReport {
        ok: 0,
        failed: 0,
        missing: 0,
        one_shot_consumed: false,
        timer_reset: 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::manager::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 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}")
            }
        };
        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");
        }
    }
    match schedule.interval_seconds {
        None => {
            // One-shot is consumed by the manual fire. (reset_timer is
            // moot here — there's no recurring cadence to re-arm.)
            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;
            }
        }
        Some(interval) if reset_timer => {
            // Recurring + operator asked to reset: re-arm the countdown
            // from now (now + interval), not along the existing cadence.
            let next = now.saturating_add(i64::try_from(interval).unwrap_or(i64::MAX));
            if let Err(e) = coord.scheduled_prompts.set_next_fire(schedule_id, next) {
                tracing::warn!(error = ?e, schedule = schedule_id, "set_next_fire after manual fire reset failed");
            } else {
                report.timer_reset = true;
            }
        }
        // Recurring without reset: cadence stays intact (additive fire).
        Some(_) => {}
    }
    Ok(report)
}

/// 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::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());
                }
            }
        }
        Err(e) => {
            tracing::warn!(error = ?e, "fire_now: container listing failed");
        }
    }
    out
}