vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Activity

hive: remove the read-only schedules page and the old schedule store

Browse source 
Scheduled prompts live at swarm level (swarm-controller's
/api/schedules, `swarmctl schedule`, the swarm UI's /schedules page).
The hive dashboard's SCH3DUL3S tab only listed the rows left in
hive-c0re's `scheduled_prompts` table so operators could copy them
over; the operator confirmed everything is migrated.

Removed:
- the SCH3DUL3S tab, its pane, schedules.js, its CSS (including the
  `.rqe-source*` rules, whose only emitter was schedules.js) and the
  hub tile's "schedules" entry
- hive-c0re's GET /api/schedules and the `scheduled_prompts` store
  (with its v1 column migration and test)
- hive-sh4re's `schedule` wire types, which only that route used
- docs describing the tab, the route and the two tables

The rebuild-queue cancel route that shared the module moves to
dashboard/rebuild_queue.rs under an OpenAPI tag of its own.

The two sqlite tables are not dropped: nothing opens them any more,
and persistence.md says they can be dropped by hand.

Closes #4916
This commit is contained in:
atlas 2026-10-12 20:29:33 +02:00 • committed by mara
commit 78412b3d3b
22 changed files with 84 additions and 979 deletions
Download patch file Download diff file Expand all files Collapse all files

7
hive-c0re/src/coordinator.rs
View file

@ -29,10 +29,6 @@ const LAST_STOPPED_RUNNING_KEY: &str = "last_stopped_running";
pub struct Coordinator {
pub broker: Arc<Broker>,
/// Scheduled-prompt rows, read-only: the dashboard lists them so the
/// operator can recreate them with `swarmctl schedule`. Nothing on
/// the hive writes or fires them.
pub scheduled_prompts: Arc<crate::scheduled_prompts::ScheduledPrompts>,
/// Full build-log capture. `lifecycle::run` /
/// `lifecycle::prebuild_toplevel` `start()` a row per attempt,
/// pipe every stdout/stderr line into it, and `finish()` it on
@ -452,8 +448,6 @@ impl Coordinator {
agent_io_weight,
} = env;
let broker = Broker::open(db_path).context("open broker")?;
let scheduled_prompts = crate::scheduled_prompts::ScheduledPrompts::open(db_path)
.context("open scheduled_prompts")?;
// BuildLogs wants a directory (it picks its own `build_logs.sqlite`
// file under it); every other opener here takes the sibling
// sqlite-file path itself. Derive the dir from `db_path`'s
@ -481,7 +475,6 @@ impl Coordinator {
let (shutdown_tx, _) = watch::channel(false);
Ok(Self {
broker: Arc::new(broker),
scheduled_prompts: Arc::new(scheduled_prompts),
build_logs,
hyperhive_flake,
hyperhive_docs_flake,

10
hive-c0re/src/dashboard/mod.rs
View file

@ -38,7 +38,7 @@ use crate::lifecycle;
(name = "meta_inputs", description = "bulk flake-input update for the meta flake"),
(name = "misc_api", description = "mark-all-read, compose, spawn-request, hive stats"),
(name = "permissions", description = "tool-group + capability assignment for agents"),
(name = "schedules", description = "read-only scheduled-prompt list + rebuild-queue cancel"),
(name = "rebuild_queue", description = "rebuild-queue cancel"),
(name = "state_files", description = "proxied reads of allow-listed per-agent state files"),
(name = "state_snapshot", description = "cold-load dashboard snapshot"),
(name = "tombstones", description = "purge of retained state for destroyed agents"),
@ -57,7 +57,7 @@ mod lifecycle_ops;
mod meta_inputs;
mod misc_api;
pub(crate) mod permissions;
mod schedules;
mod rebuild_queue;
mod state_files;
mod state_snapshot;
mod tombstones;
@ -132,8 +132,7 @@ fn app(coord: Arc<Coordinator>) -> axum::Router {
.routes(routes!(permissions::post_permissions))
.routes(routes!(permissions::get_stale_permissions))
.routes(routes!(permissions::delete_agent_permissions))
.routes(routes!(schedules::api_schedules))
.routes(routes!(schedules::post_rebuild_queue_cancel))
.routes(routes!(rebuild_queue::post_rebuild_queue_cancel))
.routes(routes!(lifecycle_ops::post_destroy))
.routes(routes!(lifecycle_ops::post_kill))
.routes(routes!(lifecycle_ops::post_restart))
@ -389,8 +388,7 @@ mod router_build_probe {
.routes(routes!(permissions::post_permissions))
.routes(routes!(permissions::get_stale_permissions))
.routes(routes!(permissions::delete_agent_permissions))
.routes(routes!(schedules::api_schedules))
.routes(routes!(schedules::post_rebuild_queue_cancel))
.routes(routes!(rebuild_queue::post_rebuild_queue_cancel))
.routes(routes!(lifecycle_ops::post_destroy))
.routes(routes!(lifecycle_ops::post_kill))
.routes(routes!(lifecycle_ops::post_restart))

49
hive-c0re/src/dashboard/rebuild_queue.rs Normal file
View file

@ -0,0 +1,49 @@
//! Rebuild-queue cancel endpoint for the dashboard.
use axum::{
extract::{Path as AxumPath, State},
response::{IntoResponse, Response},
};
use super::AppState;
#[derive(serde::Serialize, utoipa::ToSchema)]
pub(super) struct CancelResultBody {
cancelled: bool,
}
/// 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 = "rebuild_queue"
)]
pub(super) async fn post_rebuild_queue_cancel(
State(state): State<AppState>,
AxumPath(id): AxumPath<u64>,
) -> 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()
}
}

190
hive-c0re/src/dashboard/schedules.rs
View file

@ -1,190 +0,0 @@
//! Scheduled-prompt list + rebuild-queue cancel endpoints for the dashboard.
//!
//! `/api/schedules` is read-only: it lists the rows still in the hive's
//! `scheduled_prompts` table so the operator can recreate them with
//! `swarmctl schedule` (`docs/swarm/schedules.md`). Nothing on the hive
//! writes or fires those rows. Also the co-located
//! `/api/rebuild-queue/{id}/cancel` endpoint.
use axum::{
extract::{Path as AxumPath, State},
response::{IntoResponse, Response},
};
use super::{AppState, error_response};
#[derive(serde::Serialize, utoipa::ToSchema)]
pub(super) struct CancelResultBody {
cancelled: bool,
}
/// Snapshot of every schedule row 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.
#[utoipa::path(
get,
path = "/api/schedules",
responses(
(status = 200, description = "every schedule, wire shape", body = Vec<serde_json::Value>),
(status = 500, description = "sqlite read failed"),
),
tag = "schedules"
)]
pub(super) async fn api_schedules(State(state): State<AppState>) -> Response {
match state.coord.scheduled_prompts.list() {
Ok(rows) => {
let mut wire: Vec<hive_sh4re::schedule::WireSchedule> =
rows.into_iter().map(schedule_to_wire).collect();
let live: std::collections::HashSet<String> = state
.coord
.containers_snapshot()
.await
.into_iter()
.map(|c| c.name)
.collect();
filter_ghost_schedule_targets(&mut wire, &live);
axum::Json(wire).into_response()
}
Err(e) => error_response(&format!("scheduled_prompts list: {e:#}")),
}
}
/// Drop schedule targets that point at agents which no longer exist, so
/// the 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). A view filter: the rows keep every target.
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(),
}
}
/// 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<AppState>,
AxumPath(id): AxumPath<u64>,
) -> 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()
}
}
#[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<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();
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());
}
}

2
hive-c0re/src/main.rs
View file

@ -45,7 +45,7 @@ mod workers;
pub(crate) use agent_config::{capabilities, limits, resource_limits, tool_groups, topology};
pub(crate) use stats::{container_stats, hive_stats, host_stats, otel_metrics, warnings};
pub(crate) use stores::{broker, build_logs, db, power, scheduled_prompts};
pub(crate) use stores::{broker, build_logs, db, power};
pub(crate) use workers::{agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets};
use coordinator::{Coordinator, HiveEnv, ServeConfig};

6
hive-c0re/src/stores/mod.rs
View file

@ -1,10 +1,8 @@
//! Sqlite-backed host-side stores (broker, read-only scheduled prompts,
//! build logs, power intent) plus the shared connection open/migration
//! helper (`db`). Each submodule is re-exported at the crate root, so
//! Sqlite-backed host-side stores (broker, build logs, power intent)
//! plus the shared connection open/migration helper (`db`). Each submodule is re-exported at the crate root, so
//! `crate::broker::…` etc. keep working unchanged.
pub mod broker;
pub mod build_logs;
pub mod db;
pub mod power;
pub mod scheduled_prompts;

7
hive-c0re/src/stores/power.rs
View file

@ -5,9 +5,8 @@
//!
//! Stored as the `agent_power` table in the coordinator DB
//! (`/var/lib/hyperhive/db/broker.sqlite`, one tiny row per agent) —
//! same one-file-many-modules pattern as `broker` /
//! `scheduled_prompts`, each with its own connection. Intent persists
//! across hive-c0re restarts; in-flight
//! same one-file-many-modules pattern as `broker`, each with its own
//! connection. Intent persists across hive-c0re restarts; in-flight
//! queue work deliberately does not. Setting `wanted` is never a
//! queued node: operator/intent actions update the row synchronously
//! at request time, then submit the DAG whose terminal `Reconcile`
@ -72,7 +71,7 @@ pub struct PowerStore {
impl PowerStore {
/// Open (a connection to) the shared coordinator DB and ensure the
/// `agent_power` table exists. `db_path` is the same sqlite file
/// the broker / scheduled-prompts stores open.
/// the broker store opens.
pub fn open(db_path: &Path) -> Result<Self> {
let conn = crate::db::open(db_path, "agent_power")?;
conn.execute_batch(SCHEMA)

225
hive-c0re/src/stores/scheduled_prompts.rs
View file

@ -1,225 +0,0 @@
//! Scheduled prompts left in the hive's sqlite: the rows the dashboard's
//! read-only schedule list shows so the operator can recreate them with
//! `swarmctl schedule` (`docs/swarm/schedules.md`). Nothing on the hive
//! writes or fires these rows.
//!
//! Schema: `docs/agent-lifecycle/persistence.md::/var/lib/hyperhive/db/broker.sqlite`
//! (the `scheduled_prompts` / `scheduled_prompt_targets` table bullets).
use std::path::Path;
use std::sync::Mutex;
use anyhow::{Context, Result};
use chrono::{DateTime, Utc};
use rusqlite::{Connection, params};
use serde::{Deserialize, Serialize};
use crate::db::Migration;
const SCHEMA: &str = r"
CREATE TABLE IF NOT EXISTS scheduled_prompts (
id INTEGER PRIMARY KEY AUTOINCREMENT,
owner TEXT NOT NULL,
body TEXT NOT NULL,
interval_seconds INTEGER,
next_fire_at_unix INTEGER NOT NULL,
created_at_unix INTEGER NOT NULL,
source TEXT NOT NULL,
cancelled_at_unix INTEGER,
description TEXT,
paused_at_unix INTEGER
);
CREATE TABLE IF NOT EXISTS scheduled_prompt_targets (
schedule_id INTEGER NOT NULL,
target TEXT NOT NULL,
cancelled_at_unix INTEGER,
last_fired_at_unix INTEGER,
last_result TEXT,
PRIMARY KEY (schedule_id, target),
FOREIGN KEY (schedule_id) REFERENCES scheduled_prompts(id) ON DELETE CASCADE
);
";
/// One scheduled-prompt row + its target set, as `list` returns it.
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Schedule {
pub id: i64,
/// `"operator"` or the name of the agent whose approved request
/// created the row.
pub owner: String,
pub body: String,
/// `None` = one-shot. `Some(n)` = recurring every `n` seconds.
pub interval_seconds: Option<u64>,
pub next_fire_at_unix: DateTime<Utc>,
pub created_at_unix: DateTime<Utc>,
pub source: ScheduleSource,
/// Set when the whole schedule was cancelled.
pub cancelled_at_unix: Option<DateTime<Utc>>,
pub description: Option<String>,
pub paused_at_unix: Option<DateTime<Utc>>,
pub targets: Vec<ScheduleTarget>,
}
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct ScheduleTarget {
pub target: String,
pub cancelled_at_unix: Option<DateTime<Utc>>,
pub last_fired_at_unix: Option<DateTime<Utc>>,
pub last_result: Option<String>,
}
#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)]
#[serde(tag = "kind", rename_all = "snake_case")]
pub enum ScheduleSource {
Operator,
Approval { id: i64 },
}
impl ScheduleSource {
fn from_db_string(s: &str) -> Self {
if let Some(rest) = s.strip_prefix("approval:")
&& let Ok(id) = rest.parse::<i64>()
{
return ScheduleSource::Approval { id };
}
// Unknown / "operator" / corrupt → operator (best the row
// can do without an unparseable-source variant).
ScheduleSource::Operator
}
}
pub struct ScheduledPrompts {
conn: Mutex<Connection>,
}
impl ScheduledPrompts {
pub fn open(path: &Path) -> Result<Self> {
let conn = crate::db::open(path, "scheduled_prompts")?;
conn.execute_batch(SCHEMA)
.context("apply scheduled_prompts schema")?;
// The per-migration `adds_column` guard skips the ALTER on a legacy
// DB that already has the column.
crate::db::apply_versioned_migrations(
&conn,
"scheduled_prompts",
&[
// v1: add paused_at_unix.
Migration {
sql: "ALTER TABLE scheduled_prompts ADD COLUMN paused_at_unix INTEGER",
adds_column: Some(("scheduled_prompts", "paused_at_unix")),
},
],
)?;
Ok(Self {
conn: Mutex::new(conn),
})
}
/// Every row, cancelled ones included, ordered by next fire time.
pub fn list(&self) -> Result<Vec<Schedule>> {
let conn = self.conn.lock().unwrap();
let mut stmt = conn.prepare(
"SELECT id, owner, body, interval_seconds, next_fire_at_unix,
created_at_unix, source, cancelled_at_unix, description,
paused_at_unix
FROM scheduled_prompts
ORDER BY next_fire_at_unix ASC, id ASC",
)?;
let rows = stmt.query_map([], row_to_schedule_header)?;
let mut out = Vec::new();
for row in rows {
let mut s = row?;
s.targets = load_targets(&conn, s.id)?;
out.push(s);
}
Ok(out)
}
}
fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result<Schedule> {
let interval: Option<i64> = row.get(3)?;
let source_str: String = row.get(6)?;
Ok(Schedule {
id: row.get(0)?,
owner: row.get(1)?,
body: row.get(2)?,
interval_seconds: interval.and_then(|i| u64::try_from(i).ok()),
next_fire_at_unix: hive_sh4re::wire_time::from_secs(row.get(4)?),
created_at_unix: hive_sh4re::wire_time::from_secs(row.get(5)?),
source: ScheduleSource::from_db_string(&source_str),
cancelled_at_unix: row
.get::<_, Option<i64>>(7)?
.map(hive_sh4re::wire_time::from_secs),
description: row.get(8)?,
paused_at_unix: row
.get::<_, Option<i64>>(9)?
.map(hive_sh4re::wire_time::from_secs),
targets: Vec::new(),
})
}
fn load_targets(conn: &Connection, schedule_id: i64) -> Result<Vec<ScheduleTarget>> {
let mut stmt = conn.prepare(
"SELECT target, cancelled_at_unix, last_fired_at_unix, last_result
FROM scheduled_prompt_targets
WHERE schedule_id = ?1
ORDER BY target ASC",
)?;
let rows = stmt.query_map(params![schedule_id], |row| {
Ok(ScheduleTarget {
target: row.get(0)?,
cancelled_at_unix: row
.get::<_, Option<i64>>(1)?
.map(hive_sh4re::wire_time::from_secs),
last_fired_at_unix: row
.get::<_, Option<i64>>(2)?
.map(hive_sh4re::wire_time::from_secs),
last_result: row.get(3)?,
})
})?;
let mut out = Vec::new();
for row in rows {
out.push(row?);
}
Ok(out)
}
#[cfg(test)]
mod tests {
use super::*;
use tempfile::TempDir;
#[test]
fn list_returns_rows_with_their_targets() {
let dir = TempDir::new().expect("tempdir");
let path = dir.path().join("schedules.sqlite");
let db = ScheduledPrompts::open(&path).expect("open");
let raw = Connection::open(&path).expect("raw open");
raw.execute_batch(
"INSERT INTO scheduled_prompts
(id, owner, body, interval_seconds, next_fire_at_unix,
created_at_unix, source, description)
VALUES (1, 'operator', 'later', NULL, 200, 10, 'operator', NULL),
(2, 'iris', 'hourly', 3600, 100, 10, 'approval:7', 'tick');
INSERT INTO scheduled_prompt_targets (schedule_id, target, cancelled_at_unix)
VALUES (1, 'bob', NULL), (2, 'iris', NULL), (2, 'alice', 50);",
)
.expect("seed");
let rows = db.list().expect("list");
assert_eq!(rows.iter().map(|s| s.id).collect::<Vec<_>>(), vec![2, 1]);
let hourly = &rows[0];
assert_eq!(hourly.interval_seconds, Some(3600));
assert_eq!(hourly.source, ScheduleSource::Approval { id: 7 });
assert_eq!(hourly.description.as_deref(), Some("tick"));
let targets: Vec<(&str, bool)> = hourly
.targets
.iter()
.map(|t| (t.target.as_str(), t.cancelled_at_unix.is_some()))
.collect();
assert_eq!(targets, vec![("alice", true), ("iris", false)]);
assert_eq!(rows[1].source, ScheduleSource::Operator);
assert_eq!(rows[1].interval_seconds, None);
}
}