//! 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 crate::scheduled_prompts_worker::FireNowReport; use super::{AppState, error_problem, error_response}; #[derive(serde::Serialize, utoipa::ToSchema)] pub(super) struct NewScheduleBody { id: i64, } #[derive(serde::Serialize, utoipa::ToSchema)] pub(super) struct CancelResultBody { cancelled: bool, } /// Snapshot of every schedule for the /// scheduled-prompts tab. /// /// Returns the wire shape directly so the frontend can render /// without an extra translation layer. // `hive_sh4re::schedule::WireSchedule` (the actual body) has no `ToSchema` — adding // one would pull `utoipa` into the wire-types crate for a single dashboard // endpoint. `serde_json::Value` placeholder; see the batch report. #[utoipa::path( get, path = "/api/schedules", responses( (status = 200, description = "every schedule, wire shape", body = Vec), (status = 500, description = "sqlite read failed"), ), tag = "schedules" )] 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:#}")), } } /// 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. #[utoipa::path( post, path = "/api/schedules", // `hive_sh4re::manager::SchedulePromptPayload` (the actual body) has no `ToSchema` — // same reasoning as the `Vec` placeholder on // `api_schedules` above. request_body(content = serde_json::Value, description = "SchedulePromptPayload wire shape"), responses( (status = 200, description = "created; body carries the new row id", body = NewScheduleBody), (status = 400, description = "no targets, empty body, or interval_seconds == 0"), (status = 500, description = "submit failed"), ), tag = "schedules" )] 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 .targets .iter() .any(|t| t == hive_sh4re::manager::OPERATOR_RECIPIENT) { return Err(ProblemDetails::from_status_code(StatusCode::BAD_REQUEST) .with_detail("operator is not a valid schedule 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::manager::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(NewScheduleBody { 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, utoipa::ToSchema)] pub(super) struct FireNowBody { #[serde(default)] reset_timer: bool, } /// 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`). #[utoipa::path( post, path = "/api/schedules/{id}/fire-now", params(("id" = i64, Path, description = "schedule row id")), request_body(content = FireNowBody, description = "optional; absent body means reset_timer = false"), responses( (status = 200, description = "fired; per-target outcome counts", body = FireNowReport), (status = 500, description = "schedule missing, cancelled, or fully drained"), ), tag = "schedules" )] 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:#}")), } } /// Drop still-queued work from the job /// queue. /// /// `id` is a **node** id. A DAG's root cancels the whole group (the scheduler /// cascades to pending descendants), which is what the dashboard's cancel /// button sends today — a DAG id *is* its root node's id. An interior node /// cancels just that branch. /// /// Refuses `Running` / terminal nodes: 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. #[utoipa::path( post, path = "/api/rebuild-queue/{id}/cancel", params(("id" = u64, Path, description = "job-queue node id (a DAG's root cancels the group)")), responses((status = 200, description = "whether the DAG was cancelled", body = CancelResultBody)), tag = "schedules" )] pub(super) async fn post_rebuild_queue_cancel( State(state): State, AxumPath(id): AxumPath, ) -> Response { if state.coord.job_queue.cancel(id) { // Any terminal side effect is the DAG's own spared tail node, which the // scheduler picks up on its next pass — nothing to fire from here. state.coord.emit_rebuild_queue_snapshot(); axum::Json(CancelResultBody { cancelled: true }).into_response() } else { axum::Json(CancelResultBody { cancelled: false }).into_response() } } #[derive(serde::Deserialize, Default, utoipa::ToSchema)] pub(super) struct CancelScheduleForm { /// `None` / absent / empty array → cancel whole schedule. #[serde(default)] targets: Option>, } #[derive(serde::Deserialize, Default, utoipa::ToSchema)] #[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) } /// 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. #[utoipa::path( patch, path = "/api/schedules/{id}", params(("id" = i64, Path, description = "schedule row id")), responses( (status = 200, description = "updated, wire shape", body = serde_json::Value), (status = 500, description = "update failed (cancelled, not found, ...)"), ), tag = "schedules" )] 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:#}")), } } /// 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. #[utoipa::path( post, path = "/api/schedules/{id}/pause", params(("id" = i64, Path, description = "schedule row id")), responses( (status = 200, description = "paused", body = String), (status = 404, description = "schedule cancelled or not found"), (status = 500, description = "pause failed"), ), tag = "schedules" )] 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:#}")) } } } } /// Resume a paused schedule. /// /// Idempotent; no-op on an already-active row. Returns 404 when the /// schedule is cancelled or not found. #[utoipa::path( post, path = "/api/schedules/{id}/resume", params(("id" = i64, Path, description = "schedule row id")), responses( (status = 200, description = "resumed", body = String), (status = 404, description = "schedule cancelled or not found"), (status = 500, description = "resume failed"), ), tag = "schedules" )] 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:#}")) } } } } /// 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. #[utoipa::path( post, path = "/api/schedules/{id}/cancel", params(("id" = i64, Path, description = "schedule row id")), responses( (status = 200, description = "cancelled (whole or partial)", body = String), (status = 500, description = "cancel failed"), ), tag = "schedules" )] 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:#}")), } }