vinzenz/hyperhive
vinzenz/hyperhive
Watch
0
0
Fork
You've already forked hyperhive
 0
Code Issues Pull requests Projects Releases Packages Wiki Activity Actions

add harness-local reminder store (#2635 inc 1)

Browse source
This commit is contained in:
damocles 2026-07-22 20:45:53 +02:00 • committed by mara
commit 92276e4e14
3 changed files with 371 additions and 0 deletions
Download patch file Download diff file Expand all files Collapse all files

1
hive-agent/src/main.rs
View file

@ -20,6 +20,7 @@ mod mcp_config;
mod paths;
mod plugins;
mod prompt;
mod reminders;
mod serve_common;
mod stats;
mod stream_enrich;

8
hive-agent/src/paths.rs
View file

@ -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

362
hive-agent/src/reminders.rs Normal file
View file

@ -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<String>,
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<Connection>,
}
impl Reminders {
/// Open (creating if needed) the reminder store at `path`.
///
/// # Errors
///
/// Propagates sqlite open / schema-apply failures.
pub fn open(path: &Path) -> Result<Self> {
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<i64> {
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<u64> {
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<Vec<Reminder>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<Vec<Reminder>> {
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::<rusqlite::Result<Vec<_>>>()?;
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<usize> {
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<ReminderStats> {
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<usize> {
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<Reminder> {
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<i64> = {
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::<rusqlite::Result<Vec<_>>>()
.unwrap()
};
assert_eq!(remaining_ids, vec![recent, still_pending]);
}
}