From 302738362a5b78400015eec0f6642c5d6fa2a668 Mon Sep 17 00:00:00 2001 From: damocles Date: Mon, 8 Jun 2026 22:34:52 +0200 Subject: [PATCH] refactor(#1456): extract dashboard schedule + rebuild-queue endpoints into dashboard/schedules.rs --- hive-c0re/src/dashboard.rs | 236 ++------------------------- hive-c0re/src/dashboard/schedules.rs | 227 ++++++++++++++++++++++++++ 2 files changed, 245 insertions(+), 218 deletions(-) create mode 100644 hive-c0re/src/dashboard/schedules.rs diff --git a/hive-c0re/src/dashboard.rs b/hive-c0re/src/dashboard.rs index 0bf62c77..76eb89ef 100644 --- a/hive-c0re/src/dashboard.rs +++ b/hive-c0re/src/dashboard.rs @@ -31,6 +31,7 @@ use crate::coordinator::Coordinator; use crate::lifecycle::{self, MANAGER_NAME}; mod permissions; +mod schedules; #[derive(Clone)] struct AppState { @@ -95,13 +96,25 @@ pub async fn serve(port: u16, coord: Arc) -> Result<()> { ) .route("/op-send", post(post_op_send)) .route("/meta-update", post(post_meta_update)) - .route("/api/schedules", get(api_schedules).post(post_schedule_new)) - .route("/api/schedules/{id}", axum::routing::patch(patch_schedule)) - .route("/api/schedules/{id}/cancel", post(post_schedule_cancel)) - .route("/api/schedules/{id}/fire-now", post(post_schedule_fire_now)) + .route( + "/api/schedules", + get(schedules::api_schedules).post(schedules::post_schedule_new), + ) + .route( + "/api/schedules/{id}", + axum::routing::patch(schedules::patch_schedule), + ) + .route( + "/api/schedules/{id}/cancel", + post(schedules::post_schedule_cancel), + ) + .route( + "/api/schedules/{id}/fire-now", + post(schedules::post_schedule_fire_now), + ) .route( "/api/rebuild-queue/{id}/cancel", - post(post_rebuild_queue_cancel), + post(schedules::post_rebuild_queue_cancel), ) .route("/dashboard/stream", get(dashboard_stream)) .route("/dashboard/history", get(dashboard_history)) @@ -2008,219 +2021,6 @@ async fn get_build_log_raw(State(state): State, AxumPath(id): AxumPath } } -/// `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. -async fn api_schedules(State(state): State) -> Response { - match state.coord.scheduled_prompts.list() { - Ok(rows) => axum::Json( - rows.into_iter() - .map(crate::manager_server::schedule_to_wire_public) - .collect::>(), - ) - .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. -async fn post_schedule_new( - State(state): State, - axum::Json(payload): axum::Json, -) -> Response { - if payload.targets.is_empty() { - return error_response("schedule must have at least one target"); - } - if payload.body.trim().is_empty() { - return error_response("schedule body must be non-empty"); - } - if let Some(0) = payload.interval_seconds { - return error_response("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(); - axum::Json(serde_json::json!({"id": id})).into_response() - } - Err(e) => error_response(&format!("schedule submit: {e:#}")), - } -} - -/// `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. Does NOT touch `next_fire_at_unix` on -/// recurring schedules (their cadence stays intact); one-shot -/// schedules are consumed (cancelled) by a manual fire — the -/// operator's intent is "send this now, the scheduled time was -/// wrong." -async fn post_schedule_fire_now( - State(state): State, - AxumPath(id): AxumPath, -) -> Response { - match crate::scheduled_prompts_worker::fire_now(&state.coord, id).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 `Queued` entry -/// from the rebuild queue. Refuses `Running` / terminal -/// entries: an in-flight rebuild 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 from Queued → -/// Cancelled, `{"cancelled": false}` when the row was Running / -/// terminal / gone. On success a fresh `RebuildQueueChanged` -/// snapshot fires so the row's state flip surfaces live. -async fn post_rebuild_queue_cancel( - State(state): State, - AxumPath(id): AxumPath, -) -> Response { - let cancelled = state.coord.rebuild_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)] -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" -)] -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. -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::manager_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}/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. -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:#}")), - } -} - async fn post_cancel_reminder( State(state): State, AxumPath(id): AxumPath, diff --git a/hive-c0re/src/dashboard/schedules.rs b/hive-c0re/src/dashboard/schedules.rs new file mode 100644 index 00000000..7415f808 --- /dev/null +++ b/hive-c0re/src/dashboard/schedules.rs @@ -0,0 +1,227 @@ +//! 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 super::{AppState, 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) => axum::Json( + rows.into_iter() + .map(crate::manager_server::schedule_to_wire_public) + .collect::>(), + ) + .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, +) -> Response { + if payload.targets.is_empty() { + return error_response("schedule must have at least one target"); + } + if payload.body.trim().is_empty() { + return error_response("schedule body must be non-empty"); + } + if let Some(0) = payload.interval_seconds { + return error_response("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(); + axum::Json(serde_json::json!({"id": id})).into_response() + } + Err(e) => error_response(&format!("schedule submit: {e:#}")), + } +} + +/// `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. Does NOT touch `next_fire_at_unix` on +/// recurring schedules (their cadence stays intact); one-shot +/// schedules are consumed (cancelled) by a manual fire — the +/// operator's intent is "send this now, the scheduled time was +/// wrong." +pub(super) async fn post_schedule_fire_now( + State(state): State, + AxumPath(id): AxumPath, +) -> Response { + match crate::scheduled_prompts_worker::fire_now(&state.coord, id).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 `Queued` entry +/// from the rebuild queue. Refuses `Running` / terminal +/// entries: an in-flight rebuild 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 from Queued → +/// Cancelled, `{"cancelled": false}` when the row was Running / +/// terminal / gone. On success a fresh `RebuildQueueChanged` +/// snapshot fires so the row's state flip surfaces live. +pub(super) async fn post_rebuild_queue_cancel( + State(state): State, + AxumPath(id): AxumPath, +) -> Response { + let cancelled = state.coord.rebuild_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::manager_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}/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:#}")), + } +}