diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index 1a92787a..44384520 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -21,6 +21,7 @@ mod stats_vacuum; mod flake_check; mod forge; mod lifecycle; +mod scheduled_prompts; mod limits; mod loose_ends; mod manager_server; diff --git a/hive-c0re/src/scheduled_prompts.rs b/hive-c0re/src/scheduled_prompts.rs new file mode 100644 index 00000000..9ed90fc6 --- /dev/null +++ b/hive-c0re/src/scheduled_prompts.rs @@ -0,0 +1,680 @@ +//! Scheduled prompts (closes #444). Persistent sqlite queue of +//! `(fire_at, targets, body)` rows that the worker fans out as +//! broker `Message`s to each target's inbox at fire time. Recurring +//! schedules carry `interval_seconds` and re-arm `next_fire_at` on +//! delivery; one-shots are reaped. +//! +//! ## Three submit paths +//! +//! - **Operator-direct** (`source = Operator`): the operator adds +//! a schedule through the dashboard form. Lands in the table +//! immediately, no approval gate. +//! - **Agent-requested** (`source = Approval { id }`): a sub-agent +//! (or the manager) submits a `RequestSchedulePrompt` through the +//! manager socket. An `ApprovalKind::SchedulePrompt` row is +//! queued; on approve, hive-c0re inserts the schedule row with +//! `source = Approval { id: approval_id }` so the audit trail +//! points back at the operator decision. +//! - **No self-target shortcut**: even agent-self schedules need +//! approval. The existing `remind` MCP tool stays the quick +//! self-wake path; this module is the bigger, multi-recipient, +//! operator-visible thing. +//! +//! ## Catch-up clamp (missed-while-down) +//! +//! When hive-c0re comes back from being down, the worker sees rows +//! whose `next_fire_at` is well in the past. For recurring rows +//! that would mean firing N delayed pulses in a row — spammy and +//! useless. Instead the worker fires ONCE per row and bumps +//! `next_fire_at` to the next interval slot ≥ `now`, recording how +//! many cycles were skipped in `last_result`. Operators see "fired +//! late, caught up from 17 skipped" instead of 17 wake-up storms. +//! +//! ## Per-target state +//! +//! `targets` is its own table so partial cancellation flips a +//! single row + so the dashboard can show last-fired / last-result +//! per recipient. Cancelling every target reaps the parent row on +//! the next worker pass. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result, bail}; +use rusqlite::{Connection, OptionalExtension, params}; +use serde::{Deserialize, Serialize}; + +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 +); +CREATE INDEX IF NOT EXISTS idx_scheduled_due + ON scheduled_prompts (next_fire_at_unix) + WHERE cancelled_at_unix IS NULL; + +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 current target set. Returned by +/// `list` / `get`; the per-target last-result blob is suitable for +/// rendering on the dashboard without a second query. +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct Schedule { + pub id: i64, + /// `"operator"` or an agent name — drives cancel-permission + /// checks. For approval-sourced rows this is the requesting + /// agent (the `Approval.agent` at submit time). + pub owner: String, + pub body: String, + /// `None` = one-shot, deleted after first fire. + /// `Some(n)` = recurring every `n` seconds. + pub interval_seconds: Option, + pub next_fire_at_unix: i64, + pub created_at_unix: i64, + pub source: ScheduleSource, + /// Set when the *entire* schedule was cancelled (all targets + /// flipped, or operator cancel-all). Worker reaps these on the + /// next pass. + pub cancelled_at_unix: Option, + pub description: Option, + pub targets: Vec, +} + +#[derive(Debug, Clone, Serialize, Deserialize)] +pub struct ScheduleTarget { + pub target: String, + pub cancelled_at_unix: Option, + pub last_fired_at_unix: Option, + pub last_result: Option, +} + +#[derive(Debug, Clone, Serialize, Deserialize, PartialEq, Eq)] +#[serde(tag = "kind", rename_all = "snake_case")] +pub enum ScheduleSource { + Operator, + Approval { id: i64 }, +} + +impl ScheduleSource { + fn to_db_string(&self) -> String { + match self { + ScheduleSource::Operator => "operator".to_owned(), + ScheduleSource::Approval { id } => format!("approval:{id}"), + } + } + + fn from_db_string(s: &str) -> Self { + if let Some(rest) = s.strip_prefix("approval:") + && let Ok(id) = rest.parse::() + { + return ScheduleSource::Approval { id }; + } + // Unknown / "operator" / corrupt → operator (best the row + // can do without an unparseable-source variant). + ScheduleSource::Operator + } +} + +/// Submission payload — everything the caller knows at insert time. +/// Used by both the operator-direct path and the approval-flow +/// path; the latter sets `source = Approval { id }`. +#[derive(Debug, Clone)] +pub struct NewSchedule { + pub owner: String, + pub targets: Vec, + pub body: String, + pub first_fire_at_unix: i64, + pub interval_seconds: Option, + pub description: Option, + pub source: ScheduleSource, +} + +pub struct ScheduledPrompts { + conn: Mutex, +} + +impl ScheduledPrompts { + pub fn open(path: &Path) -> Result { + if let Some(parent) = path.parent() { + std::fs::create_dir_all(parent).with_context(|| { + format!("create scheduled_prompts db parent {}", parent.display()) + })?; + } + let conn = Connection::open(path) + .with_context(|| format!("open scheduled_prompts db {}", path.display()))?; + // Required for ON DELETE CASCADE to actually fire — sqlite + // ships with FKs disabled per connection by default. + conn.execute_batch("PRAGMA foreign_keys = ON;") + .context("enable foreign keys")?; + conn.execute_batch(SCHEMA) + .context("apply scheduled_prompts schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Insert a new schedule. Returns the new id. Empty `targets` is + /// rejected — a schedule with no recipients would silently + /// never fan out, masking caller bugs. + pub fn submit(&self, new: NewSchedule) -> Result { + if new.targets.is_empty() { + bail!("schedule must have at least one target"); + } + let mut conn = self.conn.lock().unwrap(); + let tx = conn.transaction()?; + tx.execute( + "INSERT INTO scheduled_prompts + (owner, body, interval_seconds, next_fire_at_unix, + created_at_unix, source, description) + VALUES (?1, ?2, ?3, ?4, ?5, ?6, ?7)", + params![ + new.owner, + new.body, + new.interval_seconds.map(i64::try_from).and_then(Result::ok), + new.first_fire_at_unix, + now_unix(), + new.source.to_db_string(), + new.description, + ], + )?; + let id = tx.last_insert_rowid(); + for target in &new.targets { + tx.execute( + "INSERT INTO scheduled_prompt_targets (schedule_id, target) + VALUES (?1, ?2)", + params![id, target], + )?; + } + tx.commit()?; + Ok(id) + } + + /// Fetch a single schedule by id (with its target rows). + /// `Ok(None)` for a non-existent / already-reaped id. + pub fn get(&self, id: i64) -> Result> { + let conn = self.conn.lock().unwrap(); + let row = conn + .query_row( + "SELECT id, owner, body, interval_seconds, next_fire_at_unix, + created_at_unix, source, cancelled_at_unix, description + FROM scheduled_prompts WHERE id = ?1", + params![id], + row_to_schedule_header, + ) + .optional()?; + let Some(mut s) = row else { + return Ok(None); + }; + s.targets = load_targets(&conn, id)?; + Ok(Some(s)) + } + + /// Every active (non-globally-cancelled) schedule in insert + /// order. Used by the dashboard list view + the cancel-auth + /// check (the latter only needs the header but list() is the + /// shared hot path). + pub fn list(&self) -> Result> { + 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 + 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) + } + + /// Pop the set of active rows that are due (`next_fire_at_unix <= now`) + /// up to `limit`. Read-only — the worker calls `mark_fired` + /// after each successful fan-out so the rows reappear on the + /// next tick when re-armed. + pub fn due(&self, now: i64, limit: u64) -> Result> { + 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 + FROM scheduled_prompts + WHERE cancelled_at_unix IS NULL AND next_fire_at_unix <= ?1 + ORDER BY next_fire_at_unix ASC, id ASC + LIMIT ?2", + )?; + let rows = stmt.query_map(params![now, limit], 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) + } + + /// Record a per-target fan-out result. `result` is "ok" or a + /// short error string; surfaces on the dashboard last-result + /// column. No-op for a target row that's already cancelled. + pub fn record_target_result( + &self, + schedule_id: i64, + target: &str, + fired_at_unix: i64, + result: &str, + ) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE scheduled_prompt_targets + SET last_fired_at_unix = ?1, last_result = ?2 + WHERE schedule_id = ?3 AND target = ?4 AND cancelled_at_unix IS NULL", + params![fired_at_unix, result, schedule_id, target], + )?; + Ok(()) + } + + /// Advance a recurring schedule's `next_fire_at` to the smallest + /// multiple-of-interval > `from`. Returns the count of skipped + /// cycles (≥ 0); the worker stamps that into the per-row + /// last_result so operators see "caught up from N missed". + /// + /// For one-shots (`interval_seconds IS NULL`) this is a no-op + /// at the SQL level; callers should `delete` them after fan-out + /// instead. + pub fn rearm(&self, id: i64, from_unix: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let row: Option<(Option, i64)> = conn + .query_row( + "SELECT interval_seconds, next_fire_at_unix + FROM scheduled_prompts WHERE id = ?1", + params![id], + |row| Ok((row.get(0)?, row.get(1)?)), + ) + .optional()?; + let Some((interval, current_next)) = row else { + return Ok(0); + }; + let Some(interval) = interval.filter(|&i| i > 0) else { + return Ok(0); + }; + // Smallest multiple-of-interval > `from_unix`. Catch-up + // semantics: when from_unix > current_next, every step in + // between gets counted as a skipped cycle. + let mut next = current_next + interval; + let mut skipped: u64 = 0; + while next <= from_unix { + next += interval; + skipped += 1; + } + conn.execute( + "UPDATE scheduled_prompts SET next_fire_at_unix = ?1 WHERE id = ?2", + params![next, id], + )?; + Ok(skipped) + } + + /// Delete a one-shot row after its single fire. Cascades the + /// target rows via the FK constraint. Idempotent on a missing + /// id. + pub fn delete(&self, id: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "DELETE FROM scheduled_prompts WHERE id = ?1", + params![id], + )?; + Ok(()) + } + + /// Cancel the entire schedule (all targets flipped + the parent + /// row marked cancelled). Idempotent; safe to call on an + /// already-cancelled row. + pub fn cancel_all(&self, id: i64) -> Result<()> { + let mut conn = self.conn.lock().unwrap(); + let now = now_unix(); + let tx = conn.transaction()?; + tx.execute( + "UPDATE scheduled_prompts + SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) + WHERE id = ?2", + params![now, id], + )?; + tx.execute( + "UPDATE scheduled_prompt_targets + SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) + WHERE schedule_id = ?2", + params![now, id], + )?; + tx.commit()?; + Ok(()) + } + + /// Cancel a subset of a schedule's targets. When the last + /// active target is cancelled, the parent row is auto-cancelled + /// too (so the worker reaps it). Unknown targets in the list + /// are silently skipped. + pub fn cancel_targets(&self, id: i64, targets: &[String]) -> Result<()> { + let mut conn = self.conn.lock().unwrap(); + let now = now_unix(); + let tx = conn.transaction()?; + for target in targets { + tx.execute( + "UPDATE scheduled_prompt_targets + SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) + WHERE schedule_id = ?2 AND target = ?3", + params![now, id, target], + )?; + } + // If every target on this schedule is now cancelled, flip + // the parent so the worker stops scanning it. + let active_targets: i64 = tx.query_row( + "SELECT COUNT(*) FROM scheduled_prompt_targets + WHERE schedule_id = ?1 AND cancelled_at_unix IS NULL", + params![id], + |row| row.get(0), + )?; + if active_targets == 0 { + tx.execute( + "UPDATE scheduled_prompts + SET cancelled_at_unix = COALESCE(cancelled_at_unix, ?1) + WHERE id = ?2", + params![now, id], + )?; + } + tx.commit()?; + Ok(()) + } + + /// Reap cancelled rows older than `older_than_unix`. Returns + /// the number of rows deleted. Called from the worker on each + /// tick so cancellations clear out of the dashboard without an + /// extra periodic vacuum task. + pub fn reap_cancelled(&self, older_than_unix: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM scheduled_prompts + WHERE cancelled_at_unix IS NOT NULL + AND cancelled_at_unix <= ?1", + params![older_than_unix], + )?; + Ok(n) + } +} + +fn row_to_schedule_header(row: &rusqlite::Row) -> rusqlite::Result { + let interval: Option = 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: row.get(4)?, + created_at_unix: row.get(5)?, + source: ScheduleSource::from_db_string(&source_str), + cancelled_at_unix: row.get(7)?, + description: row.get(8)?, + targets: Vec::new(), + }) +} + +fn load_targets(conn: &Connection, schedule_id: i64) -> Result> { + 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(1)?, + last_fired_at_unix: row.get(2)?, + last_result: row.get(3)?, + }) + })?; + let mut out = Vec::new(); + for row in rows { + out.push(row?); + } + Ok(out) +} + +fn now_unix() -> i64 { + std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .ok() + .and_then(|d| i64::try_from(d.as_secs()).ok()) + .unwrap_or(0) +} + +#[cfg(test)] +mod tests { + use super::*; + use tempfile::TempDir; + + fn open() -> (TempDir, ScheduledPrompts) { + let dir = TempDir::new().expect("tempdir"); + let path = dir.path().join("schedules.sqlite"); + let db = ScheduledPrompts::open(&path).expect("open"); + (dir, db) + } + + fn submit_one_shot(db: &ScheduledPrompts, fire_at: i64, targets: &[&str]) -> i64 { + db.submit(NewSchedule { + owner: "operator".into(), + targets: targets.iter().map(|t| (*t).to_owned()).collect(), + body: "wake".into(), + first_fire_at_unix: fire_at, + interval_seconds: None, + description: None, + source: ScheduleSource::Operator, + }) + .expect("submit") + } + + #[test] + fn submit_and_get_round_trips_targets() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice", "bob"]); + let s = db.get(id).expect("get").expect("present"); + assert_eq!(s.id, id); + assert_eq!(s.targets.len(), 2); + assert_eq!(s.targets[0].target, "alice"); + assert_eq!(s.targets[1].target, "bob"); + assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_none())); + } + + #[test] + fn submit_rejects_empty_targets() { + let (_dir, db) = open(); + let err = db + .submit(NewSchedule { + owner: "operator".into(), + targets: Vec::new(), + body: "wake".into(), + first_fire_at_unix: 100, + interval_seconds: None, + description: None, + source: ScheduleSource::Operator, + }) + .unwrap_err(); + assert!(format!("{err:#}").contains("at least one target")); + } + + #[test] + fn due_returns_only_past_active_rows() { + let (_dir, db) = open(); + let _future = submit_one_shot(&db, 1000, &["alice"]); + let past = submit_one_shot(&db, 50, &["alice"]); + let cancelled_past = submit_one_shot(&db, 50, &["bob"]); + db.cancel_all(cancelled_past).expect("cancel"); + let due = db.due(100, 10).expect("due"); + assert_eq!(due.len(), 1); + assert_eq!(due[0].id, past); + } + + #[test] + fn rearm_handles_catch_up() { + let (_dir, db) = open(); + // Recurring every 60s, last fire at t=100. + let id = db + .submit(NewSchedule { + owner: "operator".into(), + targets: vec!["alice".into()], + body: "wake".into(), + first_fire_at_unix: 100, + interval_seconds: Some(60), + description: None, + source: ScheduleSource::Operator, + }) + .expect("submit"); + // Worker comes back at t=400 — 5 missed cycles (160, 220, + // 280, 340, 400) → next should be 460, skipped = 5. + let skipped = db.rearm(id, 400).expect("rearm"); + assert_eq!(skipped, 5); + let s = db.get(id).expect("get").expect("present"); + assert_eq!(s.next_fire_at_unix, 460); + } + + #[test] + fn rearm_advances_one_step_when_caught_up() { + let (_dir, db) = open(); + let id = db + .submit(NewSchedule { + owner: "operator".into(), + targets: vec!["alice".into()], + body: "wake".into(), + first_fire_at_unix: 100, + interval_seconds: Some(60), + description: None, + source: ScheduleSource::Operator, + }) + .expect("submit"); + // Worker fires right at t=100 — single advance to t=160. + let skipped = db.rearm(id, 100).expect("rearm"); + assert_eq!(skipped, 0); + let s = db.get(id).expect("get").expect("present"); + assert_eq!(s.next_fire_at_unix, 160); + } + + #[test] + fn rearm_is_a_no_op_for_one_shots() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice"]); + let skipped = db.rearm(id, 1000).expect("rearm"); + assert_eq!(skipped, 0); + // next_fire_at unchanged — one-shots are reaped via delete(). + let s = db.get(id).expect("get").expect("present"); + assert_eq!(s.next_fire_at_unix, 100); + } + + #[test] + fn cancel_targets_auto_cancels_parent_when_last_drops() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice", "bob"]); + db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); + let s = db.get(id).expect("get").expect("present"); + // Parent still active (bob remains). + assert!(s.cancelled_at_unix.is_none()); + db.cancel_targets(id, &["bob".to_owned()]).expect("cancel"); + let s = db.get(id).expect("get").expect("present"); + // Parent auto-cancels once every target is gone. + assert!(s.cancelled_at_unix.is_some()); + assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some())); + } + + #[test] + fn cancel_all_flips_parent_and_targets() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice", "bob"]); + db.cancel_all(id).expect("cancel"); + let s = db.get(id).expect("get").expect("present"); + assert!(s.cancelled_at_unix.is_some()); + assert!(s.targets.iter().all(|t| t.cancelled_at_unix.is_some())); + } + + #[test] + fn delete_cascades_targets() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice", "bob"]); + db.delete(id).expect("delete"); + assert!(db.get(id).expect("get").is_none()); + } + + #[test] + fn record_target_result_skips_cancelled_targets() { + let (_dir, db) = open(); + let id = submit_one_shot(&db, 100, &["alice", "bob"]); + db.cancel_targets(id, &["alice".to_owned()]).expect("cancel"); + db.record_target_result(id, "alice", 200, "ok") + .expect("record alice"); + db.record_target_result(id, "bob", 200, "ok") + .expect("record bob"); + let s = db.get(id).expect("get").expect("present"); + let alice = s.targets.iter().find(|t| t.target == "alice").unwrap(); + let bob = s.targets.iter().find(|t| t.target == "bob").unwrap(); + // Cancelled targets do NOT get last-result writes. + assert!(alice.last_fired_at_unix.is_none()); + assert!(alice.last_result.is_none()); + assert_eq!(bob.last_fired_at_unix, Some(200)); + assert_eq!(bob.last_result.as_deref(), Some("ok")); + } + + #[test] + fn approval_source_round_trips() { + let (_dir, db) = open(); + let id = db + .submit(NewSchedule { + owner: "manager".into(), + targets: vec!["alice".into()], + body: "wake".into(), + first_fire_at_unix: 100, + interval_seconds: None, + description: None, + source: ScheduleSource::Approval { id: 42 }, + }) + .expect("submit"); + let s = db.get(id).expect("get").expect("present"); + assert_eq!(s.source, ScheduleSource::Approval { id: 42 }); + } + + #[test] + fn reap_cancelled_removes_old_rows_only() { + let (_dir, db) = open(); + let stale = submit_one_shot(&db, 100, &["alice"]); + let recent = submit_one_shot(&db, 100, &["bob"]); + db.cancel_all(stale).expect("cancel stale"); + // Manually backdate the stale row. + { + let conn = db.conn.lock().unwrap(); + conn.execute( + "UPDATE scheduled_prompts SET cancelled_at_unix = ?1 WHERE id = ?2", + params![10_i64, stale], + ) + .expect("backdate"); + } + db.cancel_all(recent).expect("cancel recent"); + let n = db.reap_cancelled(100).expect("reap"); + assert_eq!(n, 1); + assert!(db.get(stale).expect("get stale").is_none()); + assert!(db.get(recent).expect("get recent").is_some()); + } +}