//! Scheduled-prompt + rebuild-queue endpoints for the dashboard. //! //! Operator-direct schedule CRUD (`/api/schedules` GET/POST, `{id}` PATCH, //! `{id}/cancel` + `{id}/fire-now`) — the operator click *is* the approval, //! so these skip the manager approval gate. Also the co-located //! `/api/rebuild-queue/{id}/cancel` endpoint. use axum::{ extract::{Path as AxumPath, State}, http::StatusCode, response::{IntoResponse, Response}, }; use problem_details::ProblemDetails; use crate::scheduled_prompts::ScheduleNotFoundOrCancelled; use super::{AppState, error_problem, error_response}; /// `GET /api/schedules` — snapshot of every schedule for the /// scheduled-prompts tab. Returns the wire shape directly /// so the frontend can render without an extra translation layer. pub(super) async fn api_schedules(State(state): State) -> Response { match state.coord.scheduled_prompts.list() { Ok(rows) => { let mut wire: Vec = rows .into_iter() .map(crate::socket_server::schedule_to_wire_public) .collect(); // Drop ghost targets (agents that no longer exist) so the // table never shows dead columns. Uses the reliable async // roster snapshot here on the request path. let live: std::collections::HashSet = state .coord .containers_snapshot() .await .into_iter() .map(|c| c.name) .collect(); crate::socket_server::filter_ghost_schedule_targets(&mut wire, &live); axum::Json(wire).into_response() } Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")), } } /// `POST /api/schedules` — operator-direct schedule creation /// (mara: "user can add them manually"). Accepts the same /// `SchedulePromptPayload` shape as the manager request flow but /// skips the approval gate — the operator click *is* the /// approval. The schedule lands directly with /// `source = Operator` and the worker picks it up at fire time. pub(super) async fn post_schedule_new( State(state): State, axum::Json(payload): axum::Json, ) -> Result { if payload.targets.is_empty() { return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) .with_detail("schedule must have at least one target")); } if payload.body.trim().is_empty() { return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) .with_detail("schedule body must be non-empty")); } if let Some(0) = payload.interval_seconds { return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) .with_detail("interval_seconds must be > 0 (use None for one-shot)")); } let new = crate::scheduled_prompts::NewSchedule { owner: hive_sh4re::OPERATOR_RECIPIENT.to_owned(), targets: payload.targets, body: payload.body, first_fire_at_unix: payload.first_fire_at_unix, interval_seconds: payload.interval_seconds, description: payload.description, source: crate::scheduled_prompts::ScheduleSource::Operator, }; match state.coord.scheduled_prompts.submit(&new) { Ok(id) => { state.coord.emit_schedules_snapshot(); Ok(axum::Json(serde_json::json!({"id": id})).into_response()) } Err(e) => Err(error_problem(&format!("schedule submit: {e:#}"))), } } /// Optional JSON body for `fire-now`. Absent / empty body ⇒ /// `reset_timer = false` (back-compat: cadence stays intact). #[derive(serde::Deserialize, Default)] pub(super) struct FireNowBody { #[serde(default)] reset_timer: bool, } /// `POST /api/schedules/{id}/fire-now` — operator-initiated /// out-of-band fire of a scheduled prompt. Runs the per-target /// fan-out once immediately and reports per-target outcome counts. /// One-shot schedules are consumed (cancelled) by a manual fire — /// the operator's intent is "send this now, the scheduled time was /// wrong." For recurring schedules the cadence stays intact unless /// the body carries `{"reset_timer": true}`, in which case the /// countdown is re-armed from now (`next_fire_at = now + interval`). pub(super) async fn post_schedule_fire_now( State(state): State, AxumPath(id): AxumPath, body: Option>, ) -> Response { let reset_timer = body.is_some_and(|axum::Json(b)| b.reset_timer); match crate::scheduled_prompts_worker::fire_now(&state.coord, id, reset_timer).await { Ok(report) => { state.coord.emit_schedules_snapshot(); axum::Json(report).into_response() } Err(e) => error_response(&format!("fire schedule {id} now: {e:#}")), } } /// `POST /api/rebuild-queue/{id}/cancel` — drop a still-fully-queued /// DAG from the job queue. Refuses `Running` / terminal DAGs: an /// in-flight node owns the agent's nix store + nixos-container update /// lock and can't be safely interrupted from the queue side. Always /// returns 200; the body is `{"cancelled": true}` on a successful /// flip to Cancelled, `{"cancelled": false}` when the DAG was /// Running / terminal / gone. On success a fresh `RebuildQueueChanged` /// snapshot fires so the state flip surfaces live. pub(super) async fn post_rebuild_queue_cancel( State(state): State, AxumPath(id): AxumPath, ) -> Response { let cancelled = state.coord.job_queue.cancel(id); if cancelled { state.coord.emit_rebuild_queue_snapshot(); axum::Json(serde_json::json!({"cancelled": true})).into_response() } else { axum::Json(serde_json::json!({"cancelled": false})).into_response() } } #[derive(serde::Deserialize, Default)] pub(super) struct CancelScheduleForm { /// `None` / absent / empty array → cancel whole schedule. #[serde(default)] targets: Option>, } #[derive(serde::Deserialize, Default)] #[allow( clippy::option_option, reason = "double-Option carries three-state PATCH semantics on the wire \ (missing key = leave alone, JSON null = clear, value = set); \ collapsing to a single Option would lose the 'clear' state" )] pub(super) struct EditScheduleForm { #[serde(default)] body: Option, /// Double-`Option` semantics on the wire: missing key = leave /// alone, explicit `null` = clear, value = set. serde's /// `deserialize_with` trick to distinguish missing from null: /// we wrap each editable field in its own helper. Simpler /// here — keep them plain `Option>` and document /// that the dashboard caller passes JSON `null` to clear. #[serde(default, deserialize_with = "deserialize_some")] description: Option>, #[serde(default, deserialize_with = "deserialize_some")] interval_seconds: Option>, #[serde(default)] next_fire_at_unix: Option, /// New targets to add. Replace-on-conflict: re-adding a /// previously-cancelled target drops the tombstone and the /// target starts fresh (operator intent on re-add = "this /// target is active again, fresh start"). #[serde(default)] targets_add: Option>, /// Targets to cancel. Same path as `cancel_targets`: /// tombstones preserve audit and the parent schedule /// auto-cancels when no active targets remain. #[serde(default)] targets_remove: Option>, } /// serde adaptor: turns missing-key into `None`, explicit-null /// into `Some(None)`, value into `Some(Some(v))`. Standard trick /// for distinguishing "field absent" from "field set to null" in /// JSON PATCH bodies. fn deserialize_some<'de, T, D>(deserializer: D) -> Result, D::Error> where T: serde::Deserialize<'de>, D: serde::Deserializer<'de>, { T::deserialize(deserializer).map(Some) } /// `PATCH /api/schedules/{id}` — partial update of an existing /// schedule. Mutable fields: `body`, `description`, /// `interval_seconds`, `next_fire_at_unix`, plus the target set /// via `targets_add` / `targets_remove`. Both target lists /// are applied in the same transaction as the scalar fields with /// removes-before-adds; re-adding a previously-removed target /// resets per-target history (fresh start); draining all targets /// auto-cancels the parent schedule. JSON body uses missing-key /// = "leave alone", explicit null = "clear" for `description` + /// `interval_seconds`. Cancelled schedules are refused — submit /// a new one instead. Returns the updated `WireSchedule` so the /// caller's post-edit refresh has the new state inline. pub(super) async fn patch_schedule( State(state): State, AxumPath(id): AxumPath, axum::Json(form): axum::Json, ) -> Response { let patch = crate::scheduled_prompts::UpdateSchedule { body: form.body, description: form.description, interval_seconds: form.interval_seconds, next_fire_at_unix: form.next_fire_at_unix, targets_add: form.targets_add, targets_remove: form.targets_remove, }; if let Err(e) = state.coord.scheduled_prompts.update(id, patch) { return error_response(&format!("edit schedule {id}: {e:#}")); } match state.coord.scheduled_prompts.get(id) { Ok(Some(s)) => { let wire = crate::socket_server::schedule_to_wire_public(s); state.coord.emit_schedules_snapshot(); axum::Json(wire).into_response() } Ok(None) => error_response(&format!("edit schedule {id}: row vanished post-update")), Err(e) => error_response(&format!("re-read schedule {id}: {e:#}")), } } /// `POST /api/schedules/{id}/pause` — pause a schedule so the worker /// skips it until explicitly resumed. Idempotent; no-op on an already- /// paused row. Returns 404 when the schedule is cancelled or not found. pub(super) async fn post_schedule_pause( State(state): State, AxumPath(id): AxumPath, ) -> Response { match state.coord.scheduled_prompts.pause(id) { Ok(()) => { state.coord.emit_schedules_snapshot(); (StatusCode::OK, "ok").into_response() } Err(e) => { if e.downcast_ref::().is_some() { (StatusCode::NOT_FOUND, format!("{e}")).into_response() } else { error_response(&format!("pause schedule {id}: {e:#}")) } } } } /// `POST /api/schedules/{id}/resume` — resume a paused schedule. /// Idempotent; no-op on an already-active row. Returns 404 when the /// schedule is cancelled or not found. pub(super) async fn post_schedule_resume( State(state): State, AxumPath(id): AxumPath, ) -> Response { match state.coord.scheduled_prompts.resume(id) { Ok(()) => { state.coord.emit_schedules_snapshot(); (StatusCode::OK, "ok").into_response() } Err(e) => { if e.downcast_ref::().is_some() { (StatusCode::NOT_FOUND, format!("{e}")).into_response() } else { error_response(&format!("resume schedule {id}: {e:#}")) } } } } /// `POST /api/schedules/{id}/cancel` — operator-side cancel /// (whole schedule when no `targets` field, partial when one is /// provided). Operator bypasses the topology check; the manager /// surface enforces it for agent callers. pub(super) async fn post_schedule_cancel( State(state): State, AxumPath(id): AxumPath, body: Option>, ) -> Response { let targets = body .and_then(|axum::Json(b)| b.targets) .filter(|t| !t.is_empty()); let result = match targets.as_deref() { Some(list) => state.coord.scheduled_prompts.cancel_targets(id, list), None => state.coord.scheduled_prompts.cancel_all(id), }; match result { Ok(()) => { state.coord.emit_schedules_snapshot(); (StatusCode::OK, "ok").into_response() } Err(e) => error_response(&format!("cancel schedule {id}: {e:#}")), } }