//! 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, 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, 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, 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, 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, pub(super) description: Option>, pub(super) interval_seconds: Option>, pub(super) next_fire_at_unix: Option, pub(super) targets_add: Option>, pub(super) targets_remove: Option>, } /// 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, 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, schedule_id: i64) -> Option { 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, ) { 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)] 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 = ["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 = 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 schedule /// handlers take `&Arc`, so there is no lighter way in; /// `open` touches nothing outside the db path it is handed. fn coordinator() -> (tempfile::TempDir, Arc) { 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) -> 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:?}" ); } }