hyperhive/hive-c0re/src/socket_server/schedules.rs

//! Scheduled-prompt request handlers (`ListSchedules` /
//! `RequestSchedulePrompt` / `CancelSchedule` / `EditSchedule` /
//! `FireScheduleNow`) and the
//! schedule-to-wire mapping reused by the dashboard
//! (`schedule_to_wire_public` / `filter_ghost_schedule_targets`).

use std::sync::Arc;

use hive_core_agent_sock::Response;

use crate::coordinator::Coordinator;

/// `ListSchedules` — snapshot every scheduled prompt. Ownership no longer
/// narrows the view: `CancelSchedule`/`EditSchedule`/`FireScheduleNow` accept
/// any requester, so filtering the listing would only hide rows the requester
/// can act on anyway.
pub(super) fn handle_list_schedules(coord: &Arc<Coordinator>, requester: &str) -> Response {
    tracing::debug!(%requester, "list schedules");
    match coord.scheduled_prompts.list() {
        Ok(schedules) => Response::Schedules {
            schedules: schedules.into_iter().map(schedule_to_wire).collect(),
        },
        Err(e) => Response::Err {
            message: format!("list scheduled prompts: {e:#}"),
        },
    }
}

/// Submit a `RequestSchedulePrompt` payload as an `ApprovalKind::SchedulePrompt`
/// row. Encodes the payload into the approval's `commit_ref` so the
/// approve handler can re-parse it without a side table. Validates
/// inputs (non-empty targets, non-empty body, sane interval) at
/// submit time — the operator should never see a malformed schedule
/// pending approval.
pub(super) fn handle_request_schedule_prompt(
    coord: &Arc<Coordinator>,
    requester: &str,
    payload: &hive_sh4re::manager::SchedulePromptPayload,
) -> Response {
    if payload.targets.is_empty() {
        return Response::Err {
            message: "schedule must have at least one target".into(),
        };
    }
    if payload
        .targets
        .iter()
        .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT)
    {
        return Response::Err {
            message: "operator is not a valid schedule target".into(),
        };
    }
    if payload.body.trim().is_empty() {
        return Response::Err {
            message: "schedule body must be non-empty".into(),
        };
    }
    if let Some(0) = payload.interval_seconds {
        return Response::Err {
            message: "interval_seconds must be > 0 (use None for one-shot)".into(),
        };
    }
    let commit_ref = match serde_json::to_string(payload) {
        Ok(s) => s,
        Err(e) => {
            return Response::Err {
                message: format!("encode SchedulePromptPayload: {e:#}"),
            };
        }
    };
    let id = match coord.approvals.submit_kind(
        requester,
        hive_sh4re::approvals::ApprovalKind::SchedulePrompt,
        &commit_ref,
        payload.description.as_deref(),
        requester,
        None,
    ) {
        Ok(id) => id,
        Err(e) => {
            return Response::Err {
                message: format!("queue schedule_prompt approval: {e:#}"),
            };
        }
    };
    tracing::info!(
        %id,
        requester,
        targets = ?payload.targets,
        first_fire_at = payload.first_fire_at_unix,
        interval = ?payload.interval_seconds,
        "schedule_prompt approval queued"
    );
    coord.emit_approval_added(crate::coordinator::ApprovalAdded {
        id,
        agent: requester,
        approval_kind: "schedule_prompt",
        sha_short: None,
        description: payload.description.clone(),
        pr_number: None,
    });
    Response::Ok
}

/// Cancel a schedule (whole or per-target). Any requester may cancel any
/// schedule, whoever owns it.
pub(super) fn handle_cancel_schedule(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
    targets: Option<&[String]>,
) -> Response {
    // Existence check only — it buys the caller a clean "not found" instead
    // of whatever the cancel itself would say about an absent row.
    if let Some(err) = require_schedule(coord, schedule_id) {
        return err;
    }
    tracing::info!(%requester, %schedule_id, "cancel schedule");
    let result = match targets {
        Some(list) if !list.is_empty() => coord
            .scheduled_prompts
            .cancel_targets(schedule_id, list)
            .map_err(|e| format!("cancel targets: {e:#}")),
        _ => coord
            .scheduled_prompts
            .cancel_all(schedule_id)
            .map_err(|e| format!("cancel all: {e:#}")),
    };
    match result {
        Ok(()) => {
            coord.emit_schedules_snapshot();
            Response::Ok
        }
        Err(message) => Response::Err { message },
    }
}

/// Dispatch a `FireScheduleNow` request. Same rule as `CancelSchedule`: any
/// requester, any schedule. The actual fan-out lives in
/// `scheduled_prompts_worker::fire_now`.
pub(super) async fn handle_fire_schedule_now(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
) -> Response {
    if let Some(err) = require_schedule(coord, schedule_id) {
        return err;
    }
    tracing::info!(%requester, %schedule_id, "fire schedule now");
    // MCP fire_schedule_now stays no-reset (cadence intact); the
    // reset-timer option is a dashboard-dialog affordance.
    match crate::scheduled_prompts_worker::fire_now(coord, schedule_id, false).await {
        Ok(_report) => {
            coord.emit_schedules_snapshot();
            Response::Ok
        }
        Err(e) => Response::Err {
            message: format!("fire schedule {schedule_id} now: {e:#}"),
        },
    }
}

/// Field-named PATCH payload for [`handle_edit_schedule`]. Every
/// field is "leave alone" when `None`; the double-`Option` fields
/// additionally distinguish clear (`Some(None)`) from set
/// (`Some(Some(v))`).
#[allow(
    clippy::option_option,
    reason = "double-Option carries three-state PATCH semantics: outer None = \
              leave alone, Some(None) = clear, Some(Some(v)) = set"
)]
pub(super) struct EditSchedulePatch {
    pub(super) body: Option<String>,
    pub(super) description: Option<Option<String>>,
    pub(super) interval_seconds: Option<Option<u64>>,
    pub(super) next_fire_at_unix: Option<i64>,
    pub(super) targets_add: Option<Vec<String>>,
    pub(super) targets_remove: Option<Vec<String>>,
}

/// Dispatch an `EditSchedule` patch. Same rule as `CancelSchedule`: any
/// requester, any schedule. Forwards the partial payload to
/// `ScheduledPrompts::update` which enforces the cancelled-row /
/// zero-interval validation. Returns `Ok` on a clean update;
/// `Err` with the underlying message on any validation
/// failure so the dashboard can surface it verbatim.
pub(super) fn handle_edit_schedule(
    coord: &Arc<Coordinator>,
    requester: &str,
    schedule_id: i64,
    patch: EditSchedulePatch,
) -> Response {
    let EditSchedulePatch {
        body,
        description,
        interval_seconds,
        next_fire_at_unix,
        targets_add,
        targets_remove,
    } = patch;
    if let Some(err) = require_schedule(coord, schedule_id) {
        return err;
    }
    tracing::info!(%requester, %schedule_id, "edit schedule");
    let patch = crate::scheduled_prompts::UpdateSchedule {
        body,
        description,
        interval_seconds,
        next_fire_at_unix,
        targets_add,
        targets_remove,
    };
    match coord.scheduled_prompts.update(schedule_id, patch) {
        Ok(()) => {
            coord.emit_schedules_snapshot();
            Response::Ok
        }
        Err(e) => Response::Err {
            message: format!("edit schedule {schedule_id}: {e:#}"),
        },
    }
}

/// Shared existence guard for the three schedule-mutating verbs: `None` when
/// the row is there, `Some(Err)` to short-circuit when it isn't or the read
/// failed. Not a permission check — any requester may act on any schedule.
fn require_schedule(coord: &Arc<Coordinator>, schedule_id: i64) -> Option<Response> {
    match coord.scheduled_prompts.get(schedule_id) {
        Ok(Some(_)) => None,
        Ok(None) => Some(Response::Err {
            message: format!("schedule {schedule_id} not found"),
        }),
        Err(e) => Some(Response::Err {
            message: format!("read schedule {schedule_id}: {e:#}"),
        }),
    }
}

/// Map a `scheduled_prompts::Schedule` to its public wire shape.
/// Field-by-field copy — the two types are intentionally identical;
/// the separation keeps hive-sh4re free of hive-c0re-internal types.
/// Public alias `schedule_to_wire_public` re-exports for
/// `dashboard.rs::api_schedules` without crossing the module
/// boundary into the socket-server file.
pub fn schedule_to_wire_public(
    s: crate::scheduled_prompts::Schedule,
) -> hive_sh4re::schedule::WireSchedule {
    schedule_to_wire(s)
}

/// Drop schedule targets that point at agents which no longer exist, so
/// the dashboard's schedule table doesn't render ghost columns for
/// destroyed agents. `live` is the set of logical agent names from the
/// last `nixos-container list` scan (stopped agents included, destroyed
/// ones absent). Applied only to the dashboard wire paths
/// (`api_schedules` + the `SchedulesChanged` SSE emit) — `list_schedules`
/// doesn't apply this filter (it still returns every live target, ghost
/// or not, on the schedules the requester is authorized to see) so an
/// agent can still see and cancel its own stale targets. This is a view
/// filter: the underlying schedule rows keep every target, so a
/// re-spawned agent's targets reappear on their own.
pub(crate) fn filter_ghost_schedule_targets(
    schedules: &mut [hive_sh4re::schedule::WireSchedule],
    live: &std::collections::HashSet<String>,
) {
    for s in schedules.iter_mut() {
        s.targets.retain(|t| live.contains(&t.target));
    }
}

fn schedule_to_wire(s: crate::scheduled_prompts::Schedule) -> hive_sh4re::schedule::WireSchedule {
    hive_sh4re::schedule::WireSchedule {
        id: s.id,
        owner: s.owner,
        body: s.body,
        interval_seconds: s.interval_seconds,
        next_fire_at_unix: s.next_fire_at_unix,
        created_at_unix: s.created_at_unix,
        source: match s.source {
            crate::scheduled_prompts::ScheduleSource::Operator => {
                hive_sh4re::schedule::WireScheduleSource::Operator
            }
            crate::scheduled_prompts::ScheduleSource::Approval { id } => {
                hive_sh4re::schedule::WireScheduleSource::Approval { id }
            }
        },
        cancelled_at_unix: s.cancelled_at_unix,
        paused_at_unix: s.paused_at_unix,
        description: s.description,
        targets: s
            .targets
            .into_iter()
            .map(|t| hive_sh4re::schedule::WireScheduleTarget {
                target: t.target,
                cancelled_at_unix: t.cancelled_at_unix,
                last_fired_at_unix: t.last_fired_at_unix,
                last_result: t.last_result,
            })
            .collect(),
    }
}

#[cfg(test)]
pub(super) mod tests {
    use super::*;

    fn target(name: &str) -> hive_sh4re::schedule::WireScheduleTarget {
        hive_sh4re::schedule::WireScheduleTarget {
            target: name.to_owned(),
            cancelled_at_unix: None,
            last_fired_at_unix: None,
            last_result: None,
        }
    }

    fn schedule(targets: &[&str]) -> hive_sh4re::schedule::WireSchedule {
        hive_sh4re::schedule::WireSchedule {
            id: 1,
            owner: "operator".to_owned(),
            body: "ping".to_owned(),
            interval_seconds: None,
            next_fire_at_unix: hive_sh4re::wire_time::from_secs(0),
            created_at_unix: hive_sh4re::wire_time::from_secs(0),
            source: hive_sh4re::schedule::WireScheduleSource::Operator,
            cancelled_at_unix: None,
            paused_at_unix: None,
            description: None,
            targets: targets.iter().map(|t| target(t)).collect(),
        }
    }

    #[test]
    fn ghost_filter_drops_dead_agents_keeps_live() {
        let live: std::collections::HashSet<String> = ["iris".to_owned(), "damocles".to_owned()]
            .into_iter()
            .collect();
        let mut schedules = vec![schedule(&["iris", "ghost", "damocles"])];
        filter_ghost_schedule_targets(&mut schedules, &live);
        let kept: Vec<&str> = schedules[0]
            .targets
            .iter()
            .map(|t| t.target.as_str())
            .collect();
        // `ghost` (destroyed) dropped; live agents kept.
        assert_eq!(kept, vec!["iris", "damocles"]);
    }

    #[test]
    fn ghost_filter_can_empty_targets_when_all_dead() {
        let live: std::collections::HashSet<String> = std::collections::HashSet::new();
        let mut schedules = vec![schedule(&["gone1", "gone2"])];
        filter_ghost_schedule_targets(&mut schedules, &live);
        assert!(schedules[0].targets.is_empty());
    }

    // -----------------------------------------------------------------------
    // The scheduling surface permits an unrelated requester
    //
    // These pin the *decision* the deleted `schedule_authorized` used to make,
    // inverted: a requester with no relation to a schedule's owner reaches the
    // handler's effect instead of a refusal. They live at the handler level
    // because the predicate, its pure `_in` form and all three wrappers are
    // gone — the handler is the lowest layer where the decision still exists
    // to be observed.
    //
    // `OWNER` and `STRANGER` are names no hive has, so no `topology.json` on
    // any machine running these tests can relate them. Any ancestry check
    // reintroduced between the two — disk-reading or pure — therefore answers
    // "unrelated", and every assertion below flips from the handler's effect
    // to a refusal.
    // -----------------------------------------------------------------------

    /// Owner of the schedule under test.
    const OWNER: &str = "schedule-owner-with-no-relations";
    /// Requester acting on it: not the owner, not the operator, and not
    /// anybody's ancestor or descendant.
    const STRANGER: &str = "requester-with-no-relations";

    /// A real `Coordinator` over a throwaway sqlite dir. The socket-server
    /// handlers take `&Arc<Coordinator>`, so there is no lighter way in;
    /// `open` touches nothing outside the db path it is handed.
    pub(in crate::socket_server) fn coordinator() -> (tempfile::TempDir, Arc<Coordinator>) {
        let dir = tempfile::tempdir().expect("tempdir");
        let coord = Coordinator::open(
            &dir.path().join("broker.sqlite"),
            crate::coordinator::HiveEnv::default(),
            crate::hive_stats::PriceTable::new(),
            1,
        )
        .expect("open coordinator");
        (dir, Arc::new(coord))
    }

    /// Insert a schedule owned by [`OWNER`], returning its id.
    fn submit_owned_by_other(coord: &Arc<Coordinator>) -> i64 {
        coord
            .scheduled_prompts
            .submit(&crate::scheduled_prompts::NewSchedule {
                owner: OWNER.to_owned(),
                targets: vec!["iris".to_owned()],
                body: "ping".to_owned(),
                first_fire_at_unix: 100,
                interval_seconds: None,
                description: None,
                source: crate::scheduled_prompts::ScheduleSource::Operator,
            })
            .expect("submit")
    }

    #[test]
    fn cancel_schedule_permits_unrelated_requester() {
        let (_dir, coord) = coordinator();
        let id = submit_owned_by_other(&coord);
        let resp = handle_cancel_schedule(&coord, STRANGER, id, None);
        assert!(
            matches!(resp, Response::Ok),
            "unrelated requester must be allowed to cancel; got {resp:?}"
        );
        // Ok alone could be a refusal that forgot to say so — check the row.
        let row = coord
            .scheduled_prompts
            .get(id)
            .expect("read back")
            .expect("row present");
        assert!(
            row.cancelled_at_unix.is_some(),
            "cancel must have taken effect, not just answered Ok"
        );
    }

    #[test]
    fn edit_schedule_permits_unrelated_requester() {
        let (_dir, coord) = coordinator();
        let id = submit_owned_by_other(&coord);
        let resp = handle_edit_schedule(
            &coord,
            STRANGER,
            id,
            EditSchedulePatch {
                body: Some("edited by a stranger".to_owned()),
                description: None,
                interval_seconds: None,
                next_fire_at_unix: None,
                targets_add: None,
                targets_remove: None,
            },
        );
        assert!(
            matches!(resp, Response::Ok),
            "unrelated requester must be allowed to edit; got {resp:?}"
        );
        let row = coord
            .scheduled_prompts
            .get(id)
            .expect("read back")
            .expect("row present");
        assert_eq!(row.body, "edited by a stranger");
    }

    #[tokio::test]
    async fn fire_schedule_now_permits_unrelated_requester() {
        let (_dir, coord) = coordinator();
        let id = submit_owned_by_other(&coord);
        let resp = handle_fire_schedule_now(&coord, STRANGER, id).await;
        // The fan-out itself is allowed to find no live target (this test has
        // no containers) — that is reported inside the report, not as an Err.
        assert!(
            matches!(resp, Response::Ok),
            "unrelated requester must be allowed to fire; got {resp:?}"
        );
    }

    #[test]
    fn list_schedules_shows_a_schedule_owned_by_someone_else() {
        let (_dir, coord) = coordinator();
        let id = submit_owned_by_other(&coord);
        let Response::Schedules { schedules } = handle_list_schedules(&coord, STRANGER) else {
            panic!("expected a Schedules response");
        };
        // The deleted filter dropped every row the requester was unrelated to,
        // which for this requester is all of them.
        assert!(
            schedules.iter().any(|s| s.id == id && s.owner == OWNER),
            "unrelated requester must see another agent's schedule; got {schedules:?}"
        );
    }
}