diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index f167da3c..d4097fb3 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -20,6 +20,7 @@ mod mcp_config; mod paths; mod plugins; mod prompt; +mod reminders; mod serve_common; mod stats; mod stream_enrich; diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index 2d7c7b99..ba71a823 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -48,6 +48,14 @@ pub fn todos_db() -> PathBuf { harness_dir().join("hyperhive-todos.sqlite") } +/// Harness-local reminder store (#2635 increment 1). Same rationale as +/// [`todos_db`] — mutable per-agent state the harness owns, kept out of +/// the append-only events sink. +#[must_use] +pub fn reminders_db() -> PathBuf { + harness_dir().join("hyperhive-reminders.sqlite") +} + /// Per-turn config dir for the regenerated claude-{mcp-config,settings, /// system-prompt} files the harness drops before each turn. Set by /// systemd via `RuntimeDirectory = "hive-config"`: a per-service runtime diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs new file mode 100644 index 00000000..a8d8d687 --- /dev/null +++ b/hive-agent/src/reminders.rs @@ -0,0 +1,362 @@ +//! Harness-local reminder store — the persistent, DB-backed half of the +//! in-container reminders migration (#2635 increment 1). Mirrors +//! `todos.rs`'s shape: one sqlite db under the harness dir, single-agent +//! scope (no `agent` column — every row belongs to this agent, unlike the +//! old c0re-side store which served every agent in the hive). +//! +//! Unlike todos (discovered passively via `get_loose_ends`), a reminder is +//! *active push*: it must fire into the turn loop at `due_at` with its own +//! message body (see `reminder_timer.rs`). Delivery is *soft*-deleted +//! (`sent_at` set, not `DELETE`) so the harness's own `/api/stats` reminder +//! rollup can still report scheduled/delivered/pending counts over a +//! trailing window (mirrors the old c0re `ReminderStats` shape exactly). A +//! periodic prune (`vacuum.rs`) reaps old delivered rows so the table +//! doesn't grow unbounded. + +use std::path::Path; +use std::sync::Mutex; + +use anyhow::{Context, Result}; +use hive_sh4re::ReminderStats; +use hive_sh4re::wire_time::now_unix; +use rusqlite::{Connection, params}; + +const SCHEMA: &str = r" +CREATE TABLE IF NOT EXISTS reminders ( + id INTEGER PRIMARY KEY AUTOINCREMENT, + message TEXT NOT NULL, + file_path TEXT, + due_at INTEGER NOT NULL, + created_at INTEGER NOT NULL, + sent_at INTEGER +); +CREATE INDEX IF NOT EXISTS idx_reminders_due ON reminders (due_at) WHERE sent_at IS NULL; +"; + +/// One reminder row. `file_path`, when set, is where the harness persists +/// the body instead of inlining it at delivery time (see the old +/// `store_remind`/`prepare_body` split this mirrors, now folded into +/// `reminder_timer.rs`). +#[derive(Debug, Clone)] +pub struct Reminder { + pub id: i64, + pub message: String, + pub file_path: Option, + pub due_at: i64, +} + +/// The harness-local reminder store. Cheap to share behind an `Arc`; the +/// inner connection is guarded by a `Mutex` (reminder ops are short sqlite +/// writes, same as `Todos`). +pub struct Reminders { + conn: Mutex, +} + +impl Reminders { + /// Open (creating if needed) the reminder store at `path`. + /// + /// # Errors + /// + /// Propagates sqlite open / schema-apply failures. + pub fn open(path: &Path) -> Result { + let conn = Connection::open(path) + .with_context(|| format!("open reminders db {}", path.display()))?; + conn.execute_batch(SCHEMA) + .context("apply reminders schema")?; + Ok(Self { + conn: Mutex::new(conn), + }) + } + + /// Store a new pending reminder, due at `due_at` (unix seconds). + /// Returns the new row id. + /// + /// # Errors + /// + /// Propagates the sqlite insert failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn store(&self, message: &str, file_path: Option<&str>, due_at: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let now = now_unix(); + conn.execute( + "INSERT INTO reminders (message, file_path, due_at, created_at, sent_at) \ + VALUES (?1, ?2, ?3, ?4, NULL)", + params![message, file_path, due_at, now], + )?; + Ok(conn.last_insert_rowid()) + } + + /// Count of currently-pending (undelivered) reminders. + /// + /// # Errors + /// + /// Propagates the sqlite query failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn count_pending(&self) -> Result { + let conn = self.conn.lock().unwrap(); + let count: i64 = conn.query_row( + "SELECT COUNT(*) FROM reminders WHERE sent_at IS NULL", + [], + |row| row.get(0), + )?; + Ok(u64::try_from(count).unwrap_or(0)) + } + + /// List pending reminders, soonest-due first — for `get_loose_ends` + /// rendering. + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn list_pending(&self) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, message, file_path, due_at \ + FROM reminders WHERE sent_at IS NULL ORDER BY due_at ASC", + )?; + let rows = stmt + .query_map([], row_to_reminder)? + .collect::>>()?; + Ok(rows) + } + + /// Pending reminders due at or before `now`, oldest-due first, capped + /// at `limit` rows — the batch the delivery timer drains per tick. + /// + /// # Errors + /// + /// Propagates the sqlite prepare / query failures. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn due(&self, now: i64, limit: u64) -> Result> { + let conn = self.conn.lock().unwrap(); + let mut stmt = conn.prepare( + "SELECT id, message, file_path, due_at \ + FROM reminders WHERE sent_at IS NULL AND due_at <= ?1 \ + ORDER BY due_at ASC LIMIT ?2", + )?; + let limit = i64::try_from(limit).unwrap_or(i64::MAX); + let rows = stmt + .query_map(params![now, limit], row_to_reminder)? + .collect::>>()?; + Ok(rows) + } + + /// Soft-delete: stamp `sent_at` so the row drops out of + /// `count_pending`/`list_pending`/`due` but survives for the rollup + /// stats until `prune_delivered_older_than` reaps it. + /// + /// # Errors + /// + /// Propagates the sqlite update failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn mark_delivered(&self, id: i64) -> Result<()> { + let conn = self.conn.lock().unwrap(); + conn.execute( + "UPDATE reminders SET sent_at = ?1 WHERE id = ?2", + params![now_unix(), id], + )?; + Ok(()) + } + + /// Cancel a still-pending reminder by id (hard delete — it never + /// fired, so there's nothing to keep for the rollup). Returns the + /// number of rows deleted (0 = unknown id, or already delivered). + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn cancel(&self, id: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL", + params![id], + )?; + Ok(n) + } + + /// Scheduled/delivered/pending counts within the trailing + /// `since_secs` window (`0` = all time) — the local equivalent of the + /// old c0re `ReminderRollup` query, same `sent_at IS NULL/NOT NULL` + /// shape. + /// + /// # Errors + /// + /// Propagates the sqlite query failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn rollup(&self, since_secs: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let cutoff = if since_secs > 0 { + now_unix().saturating_sub(since_secs) + } else { + i64::MIN + }; + let scheduled: i64 = conn.query_row( + "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1", + params![cutoff], + |row| row.get(0), + )?; + let delivered: i64 = conn.query_row( + "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NOT NULL", + params![cutoff], + |row| row.get(0), + )?; + let pending: i64 = conn.query_row( + "SELECT COUNT(*) FROM reminders WHERE created_at >= ?1 AND sent_at IS NULL", + params![cutoff], + |row| row.get(0), + )?; + Ok(ReminderStats { + scheduled: u64::try_from(scheduled).unwrap_or(0), + delivered: u64::try_from(delivered).unwrap_or(0), + pending: u64::try_from(pending).unwrap_or(0), + }) + } + + /// Reap delivered rows older than `cutoff` (unix seconds) — called + /// from `vacuum.rs`'s periodic sweep so the table doesn't grow + /// unbounded. Returns the number of rows deleted. + /// + /// # Errors + /// + /// Propagates the sqlite delete failure. + /// + /// # Panics + /// + /// Panics if the connection mutex is poisoned. + pub fn prune_delivered_older_than(&self, cutoff: i64) -> Result { + let conn = self.conn.lock().unwrap(); + let n = conn.execute( + "DELETE FROM reminders WHERE sent_at IS NOT NULL AND sent_at < ?1", + params![cutoff], + )?; + Ok(n) + } +} + +fn row_to_reminder(row: &rusqlite::Row) -> rusqlite::Result { + Ok(Reminder { + id: row.get(0)?, + message: row.get(1)?, + file_path: row.get(2)?, + due_at: row.get(3)?, + }) +} + +#[cfg(test)] +mod tests { + use super::*; + + // Return the `TempDir` alongside the store so it outlives the test — + // dropping it early deletes the dir and SQLite fails with + // `SQLITE_READONLY_DBMOVED`. + fn store() -> (tempfile::TempDir, Reminders) { + let dir = tempfile::tempdir().unwrap(); + let db = Reminders::open(&dir.path().join("reminders.sqlite")).unwrap(); + (dir, db) + } + + #[test] + fn store_and_list_pending() { + let (_dir, s) = store(); + let id = s.store("check on x", None, 1000).unwrap(); + let pending = s.list_pending().unwrap(); + assert_eq!(pending.len(), 1); + assert_eq!(pending[0].id, id); + assert_eq!(pending[0].due_at, 1000); + assert_eq!(s.count_pending().unwrap(), 1); + } + + #[test] + fn due_filters_by_time_and_limit() { + let (_dir, s) = store(); + s.store("a", None, 100).unwrap(); + s.store("b", None, 200).unwrap(); + s.store("c", None, 300).unwrap(); + let due = s.due(200, 10).unwrap(); + assert_eq!(due.len(), 2, "only a and b are due by t=200"); + let capped = s.due(300, 1).unwrap(); + assert_eq!(capped.len(), 1, "limit caps the batch"); + } + + #[test] + fn mark_delivered_removes_from_pending_but_counts_in_rollup() { + let (_dir, s) = store(); + let id = s.store("a", None, 100).unwrap(); + s.mark_delivered(id).unwrap(); + assert_eq!(s.count_pending().unwrap(), 0); + assert!(s.list_pending().unwrap().is_empty()); + let rollup = s.rollup(0).unwrap(); + assert_eq!(rollup.scheduled, 1); + assert_eq!(rollup.delivered, 1); + assert_eq!(rollup.pending, 0); + } + + #[test] + fn cancel_only_removes_pending() { + let (_dir, s) = store(); + let pending_id = s.store("a", None, 100).unwrap(); + let delivered_id = s.store("b", None, 100).unwrap(); + s.mark_delivered(delivered_id).unwrap(); + assert_eq!(s.cancel(pending_id).unwrap(), 1); + assert_eq!( + s.cancel(delivered_id).unwrap(), + 0, + "already-delivered rows aren't cancellable" + ); + assert_eq!(s.cancel(999).unwrap(), 0, "unknown id is a no-op"); + } + + #[test] + fn prune_reaps_only_old_delivered_rows() { + let (_dir, s) = store(); + let old = s.store("old", None, 100).unwrap(); + let recent = s.store("recent", None, 100).unwrap(); + let still_pending = s.store("pending", None, 100).unwrap(); + s.mark_delivered(old).unwrap(); + s.mark_delivered(recent).unwrap(); + // Backdate `old`'s sent_at directly so the cutoff test is deterministic. + { + let conn = s.conn.lock().unwrap(); + conn.execute( + "UPDATE reminders SET sent_at = 1 WHERE id = ?1", + params![old], + ) + .unwrap(); + } + let cutoff = now_unix() - 10; + let n = s.prune_delivered_older_than(cutoff).unwrap(); + assert_eq!(n, 1, "only the backdated row is older than cutoff"); + let remaining_ids: Vec = { + let conn = s.conn.lock().unwrap(); + let mut stmt = conn.prepare("SELECT id FROM reminders ORDER BY id").unwrap(); + stmt.query_map([], |row| row.get(0)) + .unwrap() + .collect::>>() + .unwrap() + }; + assert_eq!(remaining_ids, vec![recent, still_pending]); + } +}