From 18d21237bf6119d662879621c74999760c99de3d Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 20:19:10 +0200 Subject: [PATCH 1/9] add reminder ops to the in-agent socket protocol (#2635 inc 1) --- hive-agent-sock/src/lib.rs | 37 ++++++++++++++++++++++++++++++------- 1 file changed, 30 insertions(+), 7 deletions(-) diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index dc7bd7a8..fc07cc52 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -1,14 +1,15 @@ //! Wire types for the *in-agent* socket, served by the hive-agent harness //! to the in-container producers (matrix / bash MCP daemons) and -//! `forge_notify`. Currently carries the loose-ends-v2 *todo* op family; -//! more in-agent request families may be added over time (the socket is -//! deliberately named for the agent, not the todos). +//! `forge_notify`. Carries the loose-ends-v2 *todo* op family plus the +//! harness-local *reminder* op family (#2635 increment 1); more in-agent +//! request families may be added over time (the socket is deliberately +//! named for the agent, not the todos). //! //! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent //! protocol on `/run/hive/mcp.sock`: this socket never leaves the -//! container. The harness owns the todo store locally and signals its own -//! turn loop directly, so hive-c0re is not in the todo path — no broker -//! round-trip, no long-poll, no marker files. +//! container. The harness owns the todo + reminder stores locally and +//! signals its own turn loop directly, so hive-c0re is not in either path — +//! no broker round-trip, no long-poll, no marker files. use serde::{Deserialize, Serialize}; @@ -58,6 +59,25 @@ pub enum Request { }, /// The agent marks one of its own todos done, by id. MarkTodoDone { id: i64 }, + /// Schedule a reminder that fires into this agent's own turn loop at + /// `timing` (harness-local — no broker round-trip). Same semantics as + /// the old broker `Remind` request. `file_path`, when set, is where the + /// harness persists an over-cap body instead of inlining it. + StoreReminder { + message: String, + timing: hive_sh4re::ReminderTiming, + #[serde(default, skip_serializing_if = "Option::is_none")] + file_path: Option, + }, + /// List this agent's pending reminders (single-agent scope — unlike + /// the old broker query, there's no cross-agent `agent` filter here). + ListReminders, + /// Cancel one of this agent's own pending reminders by id, before it + /// fires. + CancelReminder { id: i64 }, + /// This agent's pending-reminder count — used by the pre-`remind` cap + /// check and the harness's own turn-stats sink. + CountPendingReminders, } /// A response on the in-agent socket. Serialised with a `kind` tag, @@ -69,8 +89,11 @@ pub enum Response { Ok, /// Op succeeded and touched `count` rows (clear / mark-done). Acked { count: u64 }, - /// `ListTodos` result. + /// `ListTodos` / `ListReminders` result (the latter wraps each row as + /// [`LooseEnd::Reminder`]). LooseEnds { loose_ends: Vec }, + /// `CountPendingReminders` result. + PendingRemindersCount { count: u64 }, /// Op failed; `message` is operator-facing. Err { message: String }, } From 92276e4e142717a5971676e8449b6263b838e89a Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 20:45:53 +0200 Subject: [PATCH 2/9] add harness-local reminder store (#2635 inc 1) --- hive-agent/src/main.rs | 1 + hive-agent/src/paths.rs | 8 + hive-agent/src/reminders.rs | 362 ++++++++++++++++++++++++++++++++++++ 3 files changed, 371 insertions(+) create mode 100644 hive-agent/src/reminders.rs 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]); + } +} From 174e277340436ca4f1e1b0dfd858f91f2e602592 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 20:50:22 +0200 Subject: [PATCH 3/9] add reminder delivery timer + main.rs wiring (#2635 inc 1) --- hive-agent/src/main.rs | 25 +++ hive-agent/src/reminder_timer.rs | 299 +++++++++++++++++++++++++++++++ 2 files changed, 324 insertions(+) create mode 100644 hive-agent/src/reminder_timer.rs diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index d4097fb3..4aa56feb 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 reminder_timer; mod reminders; mod serve_common; mod stats; @@ -233,6 +234,11 @@ enum RecvOutcome { LocalTodo, } +/// Reminder fires are pushed as a *real* `DeliveredMessage` (unlike +/// `LocalTodo`'s synthetic hint) since the body/id is per-row data the +/// producer (`reminder_timer`) already resolved — so the select arm +/// wraps it straight into `RecvOutcome::Message`, no dedicated variant. + /// Wire surface abstraction. `AgentSurface` is the only impl — the trait /// exists to keep the turn loop generic and testable. Every function that /// talks to the broker goes through this so there are zero hard-coded @@ -474,6 +480,22 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); } } + // Harness-local reminders (#2635 increment 1): a due reminder fires + // straight into an mpsc channel the serve loop races against the + // broker long-poll (unlike todos, a fire carries real per-row data, + // so a bare `Notify` doesn't fit — see `reminder_timer` docs). + // Best-effort like the todo store above: `reminder_timer::run` parks + // forever (never sends) instead of exiting when the store can't + // open, so the channel never observes a "closed" state. + let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel(); + let reminder_store = match reminders::Reminders::open(&paths::reminders_db()) { + Ok(store) => Some(Arc::new(store)), + Err(e) => { + tracing::error!(error = ?e, "open reminders db failed — reminder delivery disabled"); + None + } + }; + tokio::spawn(reminder_timer::run(reminder_store, reminder_tx)); if matches!(initial, LoginState::NeedsLogin) { login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; } else { @@ -491,6 +513,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { stats, &files, todo_wake, + reminder_rx, ) .await } @@ -513,6 +536,7 @@ async fn serve_loop( stats: Option, files: &turn::TurnFiles, todo_wake: Arc, + mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver, ) -> Result<()> { tracing::info!(socket = %socket.display(), "harness serve"); S::requeue_inflight(socket).await; @@ -537,6 +561,7 @@ async fn serve_loop( biased; o = S::recv_next(socket) => o, () = todo_wake.notified() => RecvOutcome::LocalTodo, + Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm), } } { RecvOutcome::Message(first) => first, diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs new file mode 100644 index 00000000..094693a5 --- /dev/null +++ b/hive-agent/src/reminder_timer.rs @@ -0,0 +1,299 @@ +//! Delivery half of the harness-local reminders store (#2635 increment 1). +//! Polls [`Reminders`] for due rows and pushes each as a +//! [`hive_sh4re::DeliveredMessage`] down an mpsc channel the serve loop +//! races against the broker long-poll — mirrors `todo_server`'s `Notify` +//! wake, but a reminder fire carries real per-row data (message/id), so a +//! bare `Notify` doesn't fit; the channel carries the finished message +//! instead. +//! +//! File-path semantics (large-body auto-save, delivery-time persist) +//! port the old `hive-c0re::workers::reminder_scheduler` / +//! `socket_server::reminders` logic, simplified: this runs *inside* the +//! agent's own container now, so the symlink-escape defense that guarded +//! `hive-c0re` (running outside the container, writing into a path an +//! agent claimed was its own) is moot — the harness IS the agent, +//! already sandboxed by the container boundary. Still keeps the cheap +//! belt-and-suspenders checks: `file_path` must resolve under +//! [`crate::paths::state_dir`] with no `..`/absolute components in the +//! relative tail. + +use std::path::{Path, PathBuf}; +use std::sync::Arc; +use std::time::Duration; + +use tokio::sync::mpsc; + +use crate::reminders::Reminders; + +/// Per-tick cap on reminders delivered — mirrors the old c0re +/// `REMINDER_BATCH_LIMIT`, same rationale (bound a deep backlog's +/// per-tick cost). +const REMINDER_BATCH_LIMIT: u64 = 100; + +/// Poll interval — matches the old c0re `POLL_INTERVAL`. +const POLL_INTERVAL: Duration = Duration::from_secs(5); + +/// Same cap the broker used to enforce on `send`/`ask`/`remind` bodies +/// ([`hive-c0re::agent_config::limits::MESSAGE_MAX_BYTES`], not +/// reachable from here — hive-agent doesn't depend on hive-c0re). +/// Duplicated rather than shared: this is the last remaining reminder +/// caller of that constant once the c0re-side store is deleted (commit 6). +const REMINDER_BODY_MAX_BYTES: usize = 4096; + +/// Maximum pending (undelivered) reminders this agent may hold at once. +/// Exceeding this makes `store` return an error so the caller backs off +/// instead of silently flooding the table. Override via +/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; `0` disables the cap. +const DEFAULT_REMIND_MAX_PENDING: u64 = 50; + +fn remind_max_pending() -> u64 { + std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT") + .ok() + .and_then(|s| s.trim().parse::().ok()) + .unwrap_or(DEFAULT_REMIND_MAX_PENDING) +} + +/// Run the delivery timer. `store: None` (reminders db failed to open at +/// boot) parks forever instead of looping — keeps `tx` alive so the +/// serve loop's receiver never observes a closed channel (which would +/// otherwise resolve immediately every iteration and busy-loop the +/// select), while cleanly disabling delivery. +pub async fn run(store: Option>, tx: mpsc::UnboundedSender) { + let Some(store) = store else { + tracing::error!("reminders db unavailable — reminder delivery disabled"); + std::future::pending::<()>().await; + return; + }; + loop { + tick(&store, &tx); + tokio::time::sleep(POLL_INTERVAL).await; + } +} + +fn tick(store: &Reminders, tx: &mpsc::UnboundedSender) { + let now = hive_sh4re::wire_time::now_unix(); + let due = match store.due(now, REMINDER_BATCH_LIMIT) { + Ok(rows) => rows, + Err(e) => { + tracing::warn!(error = ?e, "failed to query due reminders"); + return; + } + }; + for r in due { + let body = prepare_body(&r.message, r.file_path.as_deref()); + let dm = hive_sh4re::DeliveredMessage { + from: "reminder".into(), + body, + id: 0, + redelivered: false, + in_reply_to: None, + }; + if tx.send(dm).is_err() { + // Receiver (serve loop) is gone — process is shutting down. + // Leave the row pending; nothing else to do here. + tracing::warn!(reminder_id = r.id, "reminder delivery channel closed"); + return; + } + if let Err(e) = store.mark_delivered(r.id) { + tracing::warn!(reminder_id = r.id, error = ?e, "failed to mark reminder delivered"); + } + } +} + +/// Store a new reminder, applying the pending cap + large-body auto-save +/// dance. Returns the new row id, or a caller-ready error string (used +/// directly as `Response::Err.message` by the in-agent socket dispatch). +pub fn store( + store: &Reminders, + message: &str, + timing: &hive_sh4re::ReminderTiming, + file_path: Option<&str>, +) -> Result { + let max = remind_max_pending(); + if max > 0 { + let pending = store.count_pending().unwrap_or(0); + if pending >= max { + return Err(format!( + "reminder rejected: already {pending} pending reminders (cap {max}). \ + Cancel some via `cancel_loose_end` or wait for them to fire before \ + scheduling more. Override the cap with `HIVE_REMIND_MAX_PENDING_PER_AGENT`." + )); + } + } + let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?; + let (stored_message, stored_path) = prepare_remind_storage(message, file_path)?; + store + .store(&stored_message, stored_path.as_deref(), due_at) + .map_err(|e| format!("failed to store reminder: {e:#}")) +} + +/// Decide what to actually persist in the reminder row (see the old +/// `prepare_remind_storage` this ports — same three outcomes: verbatim +/// under the cap, auto-saved-to-a-generated-path over cap with no +/// caller path, or auto-saved-to-the-caller's-path over cap). +fn prepare_remind_storage( + message: &str, + file_path: Option<&str>, +) -> Result<(String, Option), String> { + if message.len() <= REMINDER_BODY_MAX_BYTES { + return Ok((message.to_owned(), file_path.map(str::to_owned))); + } + let req_path = match file_path { + Some(p) => p.to_owned(), + None => auto_reminder_path(), + }; + let path = resolve_state_path(&req_path) + .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; + write_payload(&path, message) + .map_err(|reason| format!("auto-save of large reminder body to `{req_path}` failed: {reason}"))?; + let hint = format!( + "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", + message.len() + ); + Ok((hint, None)) +} + +/// Generate a fresh auto-save path under this agent's own +/// `state/reminders/` dir. +fn auto_reminder_path() -> String { + let ts_ns = std::time::SystemTime::now() + .duration_since(std::time::UNIX_EPOCH) + .map_or(0, |d| d.as_nanos()); + crate::paths::state_dir() + .join("reminders") + .join(format!("auto-{ts_ns}.md")) + .to_string_lossy() + .into_owned() +} + +/// Build the delivered body for a due reminder: verbatim when +/// `file_path` is unset, otherwise persist to that path and return a +/// short pointer (falling back to inline delivery with a warning note +/// on any rejection/write failure, so the reminder still fires). +fn prepare_body(message: &str, file_path: Option<&str>) -> String { + let Some(req_path) = file_path else { + return message.to_owned(); + }; + let path = match resolve_state_path(req_path) { + Ok(p) => p, + Err(reason) => return inline_fallback(req_path, &format!("rejected: {reason}"), message), + }; + match write_payload(&path, message) { + Ok(()) => format!( + "reminder body persisted to `{req_path}` ({} bytes); read with your filesystem tools", + message.len() + ), + Err(reason) => inline_fallback(req_path, &reason, message), + } +} + +fn inline_fallback(req_path: &str, reason: &str, message: &str) -> String { + format!("[reminder file_path '{req_path}' {reason}; delivering body inline]\n\n{message}") +} + +/// Validate `req_path` is absolute, lives under this agent's own +/// `state_dir()`, and its relative tail has no traversal / absolute +/// components. Returns the (already-real, no host/container +/// translation needed in-container) path. +fn resolve_state_path(req_path: &str) -> Result { + let base = crate::paths::state_dir(); + let path = Path::new(req_path); + if !path.is_absolute() { + return Err(format!("must be absolute (got `{req_path}`)")); + } + let Ok(rel) = path.strip_prefix(&base) else { + return Err(format!( + "must live under `{}` (got `{req_path}`)", + base.display() + )); + }; + if rel.as_os_str().is_empty() { + return Err("file_path must include a filename, not just the state dir".to_owned()); + } + for comp in rel.components() { + match comp { + std::path::Component::Normal(_) => {} + other => { + return Err(format!( + "path component `{other:?}` not allowed (no traversal / absolute / root)" + )); + } + } + } + Ok(path.to_path_buf()) +} + +/// Write `message` to `path`, creating parent dirs as needed. +fn write_payload(path: &Path, message: &str) -> Result<(), String> { + let Some(parent) = path.parent() else { + return Err("internal: path has no parent".to_owned()); + }; + std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?; + std::fs::write(path, message).map_err(|e| format!("write failed: {e}")) +} + +/// Resolve the `due_at` unix timestamp for a `StoreReminder` request. +fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result { + use hive_sh4re::ReminderTiming; + match timing { + ReminderTiming::InSeconds { seconds } => { + let now = std::time::SystemTime::now(); + let future = now + .checked_add(std::time::Duration::from_secs(*seconds)) + .ok_or_else(|| { + anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range") + })?; + let duration = future + .duration_since(std::time::UNIX_EPOCH) + .map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?; + i64::try_from(duration.as_secs()) + .map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}")) + } + ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp), + } +} + +#[cfg(test)] +mod tests { + use super::*; + + #[test] + fn resolve_state_path_rejects_non_absolute() { + assert!(resolve_state_path("relative.md").is_err()); + } + + #[test] + fn resolve_due_at_in_seconds_is_close_to_now_plus_n() { + let due = resolve_due_at(&hive_sh4re::ReminderTiming::InSeconds { seconds: 60 }).unwrap(); + let now = hive_sh4re::wire_time::now_unix(); + assert!((due - now - 60).abs() <= 2, "due={due} now={now}"); + } + + #[test] + fn resolve_due_at_at_passes_through() { + let due = resolve_due_at(&hive_sh4re::ReminderTiming::At { + unix_timestamp: 123_456, + }) + .unwrap(); + assert_eq!(due, 123_456); + } + + #[test] + fn prepare_remind_storage_passthrough_under_cap() { + let (msg, fp) = prepare_remind_storage("small body", None).unwrap(); + assert_eq!(msg, "small body"); + assert_eq!(fp, None); + } + + #[test] + fn prepare_body_passthrough_when_no_file_path() { + assert_eq!(prepare_body("hello world", None), "hello world"); + } + + #[test] + fn prepare_body_falls_back_inline_on_bad_path() { + let s = prepare_body("payload", Some("/etc/passwd")); + assert!(s.starts_with("[reminder file_path '/etc/passwd' rejected:")); + assert!(s.contains("payload")); + } +} From ecd9030305472f46093ddd9ecbdd39e0e3af67aa Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 21:01:24 +0200 Subject: [PATCH 4/9] wire reminder ops into the in-agent socket dispatch (#2635 inc 1) --- hive-agent-sock/src/lib.rs | 7 ++ hive-agent/src/main.rs | 65 ++++++++++-------- hive-agent/src/reminder_timer.rs | 10 ++- hive-agent/src/reminders.rs | 10 ++- hive-agent/src/todo_server.rs | 113 +++++++++++++++++++++++++++---- hive-agent/src/vacuum.rs | 22 ++++++ 6 files changed, 180 insertions(+), 47 deletions(-) diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index fc07cc52..cf1f503c 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -78,6 +78,11 @@ pub enum Request { /// This agent's pending-reminder count — used by the pre-`remind` cap /// check and the harness's own turn-stats sink. CountPendingReminders, + /// Scheduled/delivered/pending reminder counts over a trailing + /// `since_secs` window (`0` = all time) — the harness's own `/api/stats` + /// reminder-activity chart data (was `hive-core-agent-sock`'s + /// `ReminderRollup` against the broker; now served from the local store). + ReminderRollup { since_secs: u64 }, } /// A response on the in-agent socket. Serialised with a `kind` tag, @@ -94,6 +99,8 @@ pub enum Response { LooseEnds { loose_ends: Vec }, /// `CountPendingReminders` result. PendingRemindersCount { count: u64 }, + /// `ReminderRollup` result. + ReminderRollup { stats: hive_sh4re::ReminderStats }, /// Op failed; `message` is operator-facing. Err { message: String }, } diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index 4aa56feb..a6556c05 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -216,6 +216,11 @@ fn graceful_stop_message() -> hive_sh4re::DeliveredMessage { /// enum so `serve_loop` can pattern-match without seeing it directly. enum RecvOutcome { /// Long-poll returned at least one message; first one is detached. + /// Also reused for a fired reminder: `reminder_timer` pushes a *real* + /// `DeliveredMessage` (unlike `LocalTodo`'s synthetic hint) since the + /// body/id is per-row data the producer already resolved, so the + /// select arm wraps it straight into this variant — no dedicated + /// `LocalReminder` variant needed. Message(hive_sh4re::DeliveredMessage), /// Long-poll timed out cleanly (empty `Messages` response). Caller /// sleeps then retries. @@ -234,11 +239,6 @@ enum RecvOutcome { LocalTodo, } -/// Reminder fires are pushed as a *real* `DeliveredMessage` (unlike -/// `LocalTodo`'s synthetic hint) since the body/id is per-row data the -/// producer (`reminder_timer`) already resolved — so the select arm -/// wraps it straight into `RecvOutcome::Message`, no dedicated variant. - /// Wire surface abstraction. `AgentSurface` is the only impl — the trait /// exists to keep the turn loop generic and testable. Every function that /// talks to the broker goes through this so there are zero hard-coded @@ -460,33 +460,15 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::error!(error = %e, "web_ui::serve exited with error"); } }); - // In-agent todo socket (loose-ends v2): the harness owns the todo store - // locally and serves the in-container producers on `HIVE_AGENT_SOCKET`. - // A new/changed upsert fires `todo_wake` so the serve loop drives a turn - // directly — no broker round-trip, no marker files. Best-effort: if the - // store can't open, the socket just isn't served. - let todo_wake = Arc::new(tokio::sync::Notify::new()); - match todos::Todos::open(&paths::todos_db()) { - Ok(store) => { - let store = Arc::new(store); - let wake = todo_wake.clone(); - tokio::spawn(async move { - if let Err(e) = todo_server::run(store, wake).await { - tracing::error!(error = %e, "in-agent todo socket exited with error"); - } - }); - } - Err(e) => { - tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); - } - } // Harness-local reminders (#2635 increment 1): a due reminder fires // straight into an mpsc channel the serve loop races against the // broker long-poll (unlike todos, a fire carries real per-row data, // so a bare `Notify` doesn't fit — see `reminder_timer` docs). - // Best-effort like the todo store above: `reminder_timer::run` parks - // forever (never sends) instead of exiting when the store can't - // open, so the channel never observes a "closed" state. + // Best-effort: `reminder_timer::run` parks forever (never sends) + // instead of exiting when the store can't open, so the channel + // never observes a "closed" state. Opened before the todo socket + // below so the same `Arc` can be handed to its request dispatch + // (reminder ops share the todo socket/listener). let (reminder_tx, reminder_rx) = tokio::sync::mpsc::unbounded_channel(); let reminder_store = match reminders::Reminders::open(&paths::reminders_db()) { Ok(store) => Some(Arc::new(store)), @@ -495,7 +477,32 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { None } }; - tokio::spawn(reminder_timer::run(reminder_store, reminder_tx)); + tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx)); + // In-agent todo socket (loose-ends v2 + #2635 reminders): the harness + // owns the todo + reminder stores locally and serves the + // in-container producers on `HIVE_AGENT_SOCKET`. A new/changed todo + // upsert fires `todo_wake` so the serve loop drives a turn directly — + // no broker round-trip, no marker files. Best-effort: if the todos + // store can't open, the whole socket isn't served (reminder ops ride + // along on the same listener, so they're gated on the same store — + // acceptable since a from-scratch harness boot either has a writable + // harness dir or doesn't). + let todo_wake = Arc::new(tokio::sync::Notify::new()); + match todos::Todos::open(&paths::todos_db()) { + Ok(store) => { + let store = Arc::new(store); + let wake = todo_wake.clone(); + let reminders = reminder_store.clone(); + tokio::spawn(async move { + if let Err(e) = todo_server::run(store, wake, reminders).await { + tracing::error!(error = %e, "in-agent todo socket exited with error"); + } + }); + } + Err(e) => { + tracing::error!(error = ?e, "open todos db failed — in-agent todo socket disabled"); + } + } if matches!(initial, LoginState::NeedsLogin) { login::wait_for_login(&claude_dir, login_state.clone(), &bus, poll_ms).await; } else { diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 094693a5..53fa1a00 100644 --- a/hive-agent/src/reminder_timer.rs +++ b/hive-agent/src/reminder_timer.rs @@ -58,7 +58,10 @@ fn remind_max_pending() -> u64 { /// serve loop's receiver never observes a closed channel (which would /// otherwise resolve immediately every iteration and busy-loop the /// select), while cleanly disabling delivery. -pub async fn run(store: Option>, tx: mpsc::UnboundedSender) { +pub async fn run( + store: Option>, + tx: mpsc::UnboundedSender, +) { let Some(store) = store else { tracing::error!("reminders db unavailable — reminder delivery disabled"); std::future::pending::<()>().await; @@ -144,8 +147,9 @@ fn prepare_remind_storage( }; let path = resolve_state_path(&req_path) .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; - write_payload(&path, message) - .map_err(|reason| format!("auto-save of large reminder body to `{req_path}` failed: {reason}"))?; + write_payload(&path, message).map_err(|reason| { + format!("auto-save of large reminder body to `{req_path}` failed: {reason}") + })?; let hint = format!( "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", message.len() diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs index a8d8d687..a84cd18c 100644 --- a/hive-agent/src/reminders.rs +++ b/hive-agent/src/reminders.rs @@ -43,6 +43,7 @@ pub struct Reminder { pub message: String, pub file_path: Option, pub due_at: i64, + pub created_at: i64, } /// The harness-local reminder store. Cheap to share behind an `Arc`; the @@ -121,7 +122,7 @@ impl Reminders { pub fn list_pending(&self) -> Result> { let conn = self.conn.lock().unwrap(); let mut stmt = conn.prepare( - "SELECT id, message, file_path, due_at \ + "SELECT id, message, file_path, due_at, created_at \ FROM reminders WHERE sent_at IS NULL ORDER BY due_at ASC", )?; let rows = stmt @@ -143,7 +144,7 @@ impl Reminders { 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 \ + "SELECT id, message, file_path, due_at, created_at \ FROM reminders WHERE sent_at IS NULL AND due_at <= ?1 \ ORDER BY due_at ASC LIMIT ?2", )?; @@ -262,6 +263,7 @@ fn row_to_reminder(row: &rusqlite::Row) -> rusqlite::Result { message: row.get(1)?, file_path: row.get(2)?, due_at: row.get(3)?, + created_at: row.get(4)?, }) } @@ -351,7 +353,9 @@ mod tests { 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(); + let mut stmt = conn + .prepare("SELECT id FROM reminders ORDER BY id") + .unwrap(); stmt.query_map([], |row| row.get(0)) .unwrap() .collect::>>() diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 7ad2382c..5ce462e4 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1,9 +1,14 @@ -//! In-agent socket server (loose-ends v2). Binds the harness-owned -//! `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` protocol to the -//! in-container producers (matrix / bash daemons, forge-notify). Todo ops -//! hit the harness-local [`Todos`] store; a new-or-changed upsert fires an -//! in-process [`Notify`] so the serve loop drives a turn — no hive-c0re -//! round-trip, no broker long-poll, no marker files. +//! In-agent socket server (loose-ends v2 + #2635 reminders). Binds the +//! harness-owned `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` +//! protocol to the in-container producers (matrix / bash daemons, +//! forge-notify) and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/ +//! `cancel_loose_end` tool impls. Todo ops hit the harness-local [`Todos`] +//! store; a new-or-changed upsert fires an in-process [`Notify`] so the +//! serve loop drives a turn. Reminder ops hit the harness-local +//! [`Reminders`] store (`None` when the store failed to open — every +//! reminder op then returns `Response::Err`); a reminder *firing* is a +//! separate path (`reminder_timer`), not driven through this socket. No +//! hive-c0re round-trip, no broker long-poll, no marker files. //! //! One request/response line per connection, matching the producers' //! existing best-effort JSON-line clients (they just change which socket @@ -19,6 +24,7 @@ use tokio::io::{AsyncBufReadExt, AsyncWriteExt, BufReader}; use tokio::net::{UnixListener, UnixStream}; use tokio::sync::Notify; +use crate::reminders::{Reminder, Reminders}; use crate::todos::{Todo, Todos}; /// Resolve the in-agent socket path from `HIVE_AGENT_SOCKET`. `None` when @@ -36,7 +42,11 @@ fn socket_path() -> Option { /// # Errors /// /// Returns an error if the socket path is set but can't be bound. -pub async fn run(store: Arc, wake: Arc) -> Result<()> { +pub async fn run( + store: Arc, + wake: Arc, + reminders: Option>, +) -> Result<()> { let Some(path) = socket_path() else { tracing::info!("HIVE_AGENT_SOCKET unset — in-agent todo socket disabled"); return Ok(()); @@ -48,8 +58,9 @@ pub async fn run(store: Arc, wake: Arc) -> Result<()> { Ok((stream, _)) => { let store = store.clone(); let wake = wake.clone(); + let reminders = reminders.clone(); tokio::spawn(async move { - if let Err(e) = handle_conn(stream, &store, &wake).await { + if let Err(e) = handle_conn(stream, &store, &wake, reminders.as_deref()).await { tracing::warn!(error = ?e, "in-agent todo connection failed"); } }); @@ -75,7 +86,12 @@ fn bind(path: &Path) -> Result { /// Handle one connection: read a single JSON request line, apply it to the /// store, write the JSON response line back. -async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result<()> { +async fn handle_conn( + stream: UnixStream, + store: &Todos, + wake: &Notify, + reminders: Option<&Reminders>, +) -> Result<()> { let (read, mut write) = stream.into_split(); let mut reader = BufReader::new(read); let mut line = String::new(); @@ -83,7 +99,7 @@ async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result return Ok(()); } let resp = match serde_json::from_str::(line.trim()) { - Ok(req) => dispatch(req, store, wake), + Ok(req) => dispatch(req, store, wake, reminders), Err(e) => Response::Err { message: format!("bad request: {e}"), }, @@ -96,8 +112,9 @@ async fn handle_conn(stream: UnixStream, store: &Todos, wake: &Notify) -> Result } /// Apply one request to the store, firing `wake` on a new/changed upsert so -/// the serve loop runs a turn. -fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response { +/// the serve loop runs a turn. `reminders` is `None` when that store +/// failed to open at boot — every reminder op then returns an `Err`. +fn dispatch(req: Request, store: &Todos, wake: &Notify, reminders: Option<&Reminders>) -> Response { match req { Request::UpsertTodo { subsystem, @@ -142,6 +159,78 @@ fn dispatch(req: Request, store: &Todos, wake: &Notify) -> Response { }, Err(e) => err(&e), }, + Request::StoreReminder { + message, + timing, + file_path, + } => match reminders { + Some(r) => { + match crate::reminder_timer::store(r, &message, &timing, file_path.as_deref()) { + Ok(_id) => Response::Ok, + Err(message) => Response::Err { message }, + } + } + None => no_reminders_store(), + }, + Request::ListReminders => match reminders { + Some(r) => match r.list_pending() { + Ok(rows) => Response::LooseEnds { + loose_ends: rows.into_iter().map(reminder_to_loose_end).collect(), + }, + Err(e) => err(&e), + }, + None => no_reminders_store(), + }, + Request::CancelReminder { id } => match reminders { + Some(r) => match r.cancel(id) { + Ok(count) => Response::Acked { + count: u64::try_from(count).unwrap_or(0), + }, + Err(e) => err(&e), + }, + None => no_reminders_store(), + }, + Request::CountPendingReminders => match reminders { + Some(r) => match r.count_pending() { + Ok(count) => Response::PendingRemindersCount { count }, + Err(e) => err(&e), + }, + None => no_reminders_store(), + }, + Request::ReminderRollup { since_secs } => match reminders { + Some(r) => { + let since_secs = i64::try_from(since_secs).unwrap_or(i64::MAX); + match r.rollup(since_secs) { + Ok(stats) => Response::ReminderRollup { stats }, + Err(e) => err(&e), + } + } + None => no_reminders_store(), + }, + } +} + +/// Shared "reminders db unavailable" response for every reminder op when +/// the store failed to open at boot (see `main.rs`'s best-effort open). +fn no_reminders_store() -> Response { + Response::Err { + message: "reminders store unavailable on this agent".to_owned(), + } +} + +/// Map a stored [`Reminder`] to a [`LooseEnd::Reminder`], deriving +/// `age_seconds` from `created_at` (mirrors the old c0re rendering — +/// "age" is how long the reminder has been *scheduled*, not how soon +/// it's due). +fn reminder_to_loose_end(r: Reminder) -> LooseEnd { + let now = hive_sh4re::wire_time::now_unix(); + let age = u64::try_from(now.saturating_sub(r.created_at)).unwrap_or(0); + LooseEnd::Reminder { + id: r.id, + owner: crate::identity::label(), + message: r.message, + due_at: hive_sh4re::wire_time::from_secs(r.due_at), + age_seconds: age, } } diff --git a/hive-agent/src/vacuum.rs b/hive-agent/src/vacuum.rs index c33ec8ad..06275f9a 100644 --- a/hive-agent/src/vacuum.rs +++ b/hive-agent/src/vacuum.rs @@ -28,6 +28,10 @@ const BASH_KEEP_SECS: i64 = 48 * 3600; /// kinds are never deleted by this sweep — they carry the semantic per-turn /// history the operator scrolls back through. const STREAM_KEEP_SECS: i64 = 14 * 24 * 3600; +/// Keep delivered (soft-deleted, `sent_at` set) reminder rows this long +/// before reaping them — same window as `STREAM_KEEP_SECS`, kept around +/// only to serve the trailing-window `ReminderRollup` stats. +const REMINDER_KEEP_SECS: i64 = 14 * 24 * 3600; /// Terminal bash-task statuses whose files are eligible for deletion. const TERMINAL_STATUSES: &[&str] = &["done", "timed_out", "interrupted"]; @@ -60,6 +64,24 @@ fn sweep_once() { Err(e) => tracing::warn!(error = ?e, "events vacuum failed"), } } + + let reminders_db = crate::paths::reminders_db(); + if reminders_db.exists() { + match vacuum_reminders(&reminders_db) { + Ok(0) => {} + Ok(n) => tracing::info!(removed = n, "reminders vacuum"), + Err(e) => tracing::warn!(error = ?e, "reminders vacuum failed"), + } + } +} + +/// Reap delivered reminder rows older than [`REMINDER_KEEP_SECS`] via the +/// typed store API (own short-lived connection — mirrors `vacuum_events`'s +/// own connection to `events.sqlite` rather than sharing the harness's live +/// `Reminders` handle). +fn vacuum_reminders(path: &Path) -> anyhow::Result { + let store = crate::reminders::Reminders::open(path)?; + store.prune_delivered_older_than(now_unix() - REMINDER_KEEP_SECS) } /// Delete eligible bash-task trios in `dir`. Returns the count of `.json` From 8b35b8c4af05078584eae9cd6c58029768bc0350 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 21:02:44 +0200 Subject: [PATCH 5/9] repoint per-agent stats reminder rollup off the broker (#2635 inc 1) --- hive-agent/src/web_ui/stats.rs | 61 ++++++++++++++++++++++------------ 1 file changed, 39 insertions(+), 22 deletions(-) diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index 8bf59747..1852e873 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -12,38 +12,55 @@ pub(super) struct StatsQuery { } pub(super) async fn api_stats( - State(state): State, + State(_state): State, axum::extract::Query(q): axum::extract::Query, ) -> axum::Json { let window = crate::stats::Window::parse(q.window.as_deref().unwrap_or("24h")); let mut snapshot = crate::stats::snapshot_default(window); - // Pass the window span to the reminder-stats RPC so the broker - // filters its counts to the same time range as the chart data. + // Pass the window span so the local reminder rollup filters its counts + // to the same time range as the chart data. let window_secs = window.span_secs(); let window_secs_u = u64::try_from(window_secs).unwrap_or(0); - snapshot.reminder_stats = fetch_reminder_stats(&state.socket, window_secs_u).await; + snapshot.reminder_stats = fetch_reminder_stats(window_secs_u).await; axum::Json(snapshot) } -/// Fetch reminder activity stats from the broker via the per-agent / manager -/// socket. Returns None on any transport / decode failure — the stats are -/// decorative, not authoritative. -async fn fetch_reminder_stats( - socket: &std::path::Path, - window_secs: u64, -) -> Option { - match super::broker_request( - socket, - &hive_core_agent_sock::Request::ReminderRollup { - since_secs: window_secs, - agent: None, - }, - ) - .await - { - Ok(hive_core_agent_sock::Response::ReminderRollup(stats)) => Some(stats), - _ => None, +/// Fetch reminder activity stats from the harness-local reminder store over +/// `HIVE_AGENT_SOCKET` (#2635 inc 1 — was a broker RPC before reminders +/// moved in-container). Returns `None` on any transport / decode failure or +/// when the socket is unset — the stats are decorative, not authoritative. +async fn fetch_reminder_stats(window_secs: u64) -> Option { + use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; + use tokio::net::UnixStream; + + let socket_path = std::env::var_os("HIVE_AGENT_SOCKET").map(std::path::PathBuf::from)?; + if !socket_path.exists() { + return None; } + tokio::time::timeout(std::time::Duration::from_secs(3), async move { + let mut stream = UnixStream::connect(&socket_path).await?; + let req = hive_agent_sock::Request::ReminderRollup { + since_secs: window_secs, + }; + let mut line = serde_json::to_string(&req)?; + line.push('\n'); + stream.write_all(line.as_bytes()).await?; + stream.flush().await?; + let mut lines = BufReader::new(stream).lines(); + let resp_line = lines + .next_line() + .await? + .ok_or_else(|| anyhow::anyhow!("agent socket closed without response"))?; + let resp: hive_agent_sock::Response = serde_json::from_str(&resp_line)?; + anyhow::Ok(match resp { + hive_agent_sock::Response::ReminderRollup { stats } => Some(stats), + _ => None, + }) + }) + .await + .ok()? + .ok() + .flatten() } /// `GET /api/todos` — snapshot of this agent's local todos (loose-ends v2). From e8d101d663019217422db0123a8d37a38985ff0e Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 21:05:15 +0200 Subject: [PATCH 6/9] repoint remind/cancel_loose_end/get_loose_ends to the local reminder socket (#2635 inc 1) --- hive-agent-mcp/src/mcp/mod.rs | 58 +++++++++++++++++++++++++------- hive-agent-mcp/src/mcp/render.rs | 16 +++++++-- 2 files changed, 59 insertions(+), 15 deletions(-) diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 0ca18172..3a4616c0 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -35,8 +35,9 @@ pub use args::{ pub use render::{annotate_retries, format_ack, format_agent_meta, format_recv}; use render::{ - format_matrix_summary, local_todos, loose_end_kind_label, mark_local_todo_done, - matrix_unread_summary, parse_loose_end_kind, render_loose_ends, reply_err, + dial_agent_socket, format_matrix_summary, local_reminders, local_todos, loose_end_kind_label, + mark_local_todo_done, matrix_unread_summary, parse_loose_end_kind, render_loose_ends, + reply_err, }; /// Write (or remove) the status file in the agent's own `state/` directory. @@ -343,6 +344,12 @@ impl AgentServer { if is_self_query && let Some(todos) = local_todos().await { loose_ends.extend(todos); } + // Merge local pending reminders (#2635 inc 1 — same self-query-only + // restriction: a manager asking for a child's loose-ends no longer + // sees the child's reminders, matching the todos precedent above). + if is_self_query && let Some(reminders) = local_reminders().await { + loose_ends.extend(reminders); + } annotate_retries(render_loose_ends(&loose_ends), retries) }) .await @@ -446,6 +453,26 @@ impl AgentServer { Err(e) => return e, }; let kind_label = loose_end_kind_label(kind); + // Reminders are harness-local (#2635 inc 1) — dial the in-agent + // socket directly instead of the broker; every other kind + // (question/approval) still lives in c0re. + if kind == hive_sh4re::CancelLooseEndKind::Reminder { + return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id }) + .await + { + Some(hive_agent_sock::Response::Acked { count }) if count > 0 => { + format!("cancelled {kind_label} {id}") + } + Some(hive_agent_sock::Response::Acked { .. }) => { + format!("cancel_loose_end failed: no pending {kind_label} {id}") + } + Some(hive_agent_sock::Response::Err { message }) => { + format!("cancel_loose_end failed: {message}") + } + Some(other) => format!("cancel_loose_end unexpected response: {other:?}"), + None => "cancel_loose_end failed: in-agent socket unavailable".to_owned(), + }; + } let (resp, retries) = self .dispatch(hive_core_agent_sock::Request::CancelLooseEnd { kind, id }) .await; @@ -516,17 +543,22 @@ impl AgentServer { (Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s }, (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; - let (resp, retries) = self - .dispatch(hive_core_agent_sock::Request::Remind { - message: args.message, - timing, - file_path: args.file_path, - }) - .await; - annotate_retries( - format_ack(resp, "remind", "reminder scheduled".to_string()), - retries, - ) + // Reminders are harness-local (#2635 inc 1) — dial the in-agent + // socket directly instead of the broker. + match dial_agent_socket(&hive_agent_sock::Request::StoreReminder { + message: args.message, + timing, + file_path: args.file_path, + }) + .await + { + Some(hive_agent_sock::Response::Ok) => "reminder scheduled".to_owned(), + Some(hive_agent_sock::Response::Err { message }) => { + format!("remind failed: {message}") + } + Some(other) => format!("remind unexpected response: {other:?}"), + None => "remind failed: in-agent socket unavailable".to_owned(), + } }) .await } diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index 8402c995..7217b873 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -304,8 +304,10 @@ pub(super) async fn matrix_unread_summary() -> Option> { /// best-effort, single-shot (no retry, unlike the broker's /// `client::request_retried`): this socket lives in the SAME container, so /// a connect failure means the harness itself isn't up, which a retry -/// won't fix within a tool call's budget. Shared by [`local_todos`] and -/// [`mark_local_todo_done`]. +/// won't fix within a tool call's budget. Shared by [`local_todos`], +/// [`mark_local_todo_done`], and [`local_reminders`]; `remind`/ +/// `cancel_loose_end`(reminder) in `mod.rs` dial it directly since their +/// happy path is a plain `Ok`/`Err`, not a `Vec` to merge. pub(super) async fn dial_agent_socket( req: &hive_agent_sock::Request, ) -> Option { @@ -335,6 +337,16 @@ pub(super) async fn local_todos() -> Option> { } } +/// Query the harness's in-agent socket for this agent's local pending +/// reminders (#2635 inc 1 — was a broker query before reminders moved +/// in-container). Same best-effort contract as [`local_todos`]. +pub(super) async fn local_reminders() -> Option> { + match dial_agent_socket(&hive_agent_sock::Request::ListReminders).await? { + hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends), + _ => None, + } +} + /// Mark one of this agent's local todos (loose-ends v2) done by id, via /// the harness's in-agent socket — reachable through `cancel_loose_end` /// kind `"todo"` so clearing a todo never has to shell out through a From e11e8294a38ed496a3f578ab60fa0fdcdc692833 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 21:14:31 +0200 Subject: [PATCH 7/9] repoint post_turn_counts + dedup web_ui stats onto todo_server::dial (#2635 inc 1) --- hive-agent/src/main.rs | 16 ++++---- hive-agent/src/todo_server.rs | 26 ++++++++++++ hive-agent/src/web_ui/stats.rs | 73 ++++++---------------------------- 3 files changed, 46 insertions(+), 69 deletions(-) diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index a6556c05..b18307cb 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -332,15 +332,13 @@ impl Surface for AgentSurface { Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), _ => None, }; - let reminders = match client::request::<_, Response>( - socket, - &Request::CountPendingReminders { agent: None }, - ) - .await - { - Ok(Response::PendingRemindersCount { count }) => Some(count), - _ => None, - }; + // Reminders are harness-local (#2635 inc 1) — dial the in-agent + // socket directly instead of the broker. + let reminders = + match todo_server::dial(&hive_agent_sock::Request::CountPendingReminders).await { + Some(hive_agent_sock::Response::PendingRemindersCount { count }) => Some(count), + _ => None, + }; (threads, reminders) } diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 5ce462e4..5d3a0ccf 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -35,6 +35,32 @@ fn socket_path() -> Option { .map(PathBuf::from) } +/// Dial this same in-agent socket from elsewhere IN THIS PROCESS — the +/// harness's own `web_ui` endpoints (`api_todos`, `/api/stats` reminder +/// rollup) and `Surface::post_turn_counts` all need read access to the +/// `Todos`/`Reminders` stores this module owns behind an `Arc` on a +/// different spawned task, and a loopback dial is simpler than threading +/// those `Arc`s through every caller. One-shot, best-effort (no retry, +/// unlike the broker client): a connect failure means the socket server +/// task itself isn't up, which a retry within one call wouldn't fix. +pub(crate) async fn dial(req: &Request) -> Option { + let path = socket_path()?; + if !path.exists() { + return None; + } + tokio::time::timeout(std::time::Duration::from_secs(3), async move { + let mut stream = UnixStream::connect(&path).await.ok()?; + let mut line = serde_json::to_string(req).ok()?; + line.push('\n'); + stream.write_all(line.as_bytes()).await.ok()?; + let mut lines = BufReader::new(stream).lines(); + let resp_line = lines.next_line().await.ok()??; + serde_json::from_str(&resp_line).ok() + }) + .await + .ok()? +} + /// Run the in-agent socket server: bind + accept loop, one request/response /// line per connection. A no-op (returns `Ok`) when `HIVE_AGENT_SOCKET` is /// unset, so a standalone harness without producers just skips it. diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index 1852e873..db3271f3 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -30,37 +30,14 @@ pub(super) async fn api_stats( /// moved in-container). Returns `None` on any transport / decode failure or /// when the socket is unset — the stats are decorative, not authoritative. async fn fetch_reminder_stats(window_secs: u64) -> Option { - use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; - use tokio::net::UnixStream; - - let socket_path = std::env::var_os("HIVE_AGENT_SOCKET").map(std::path::PathBuf::from)?; - if !socket_path.exists() { - return None; - } - tokio::time::timeout(std::time::Duration::from_secs(3), async move { - let mut stream = UnixStream::connect(&socket_path).await?; - let req = hive_agent_sock::Request::ReminderRollup { - since_secs: window_secs, - }; - let mut line = serde_json::to_string(&req)?; - line.push('\n'); - stream.write_all(line.as_bytes()).await?; - stream.flush().await?; - let mut lines = BufReader::new(stream).lines(); - let resp_line = lines - .next_line() - .await? - .ok_or_else(|| anyhow::anyhow!("agent socket closed without response"))?; - let resp: hive_agent_sock::Response = serde_json::from_str(&resp_line)?; - anyhow::Ok(match resp { - hive_agent_sock::Response::ReminderRollup { stats } => Some(stats), - _ => None, - }) + match crate::todo_server::dial(&hive_agent_sock::Request::ReminderRollup { + since_secs: window_secs, }) - .await - .ok()? - .ok() - .flatten() + .await? + { + hive_agent_sock::Response::ReminderRollup { stats } => Some(stats), + _ => None, + } } /// `GET /api/todos` — snapshot of this agent's local todos (loose-ends v2). @@ -70,36 +47,12 @@ async fn fetch_reminder_stats(window_secs: u64) -> Option Response { - use tokio::io::{AsyncBufReadExt as _, AsyncWriteExt as _, BufReader}; - use tokio::net::UnixStream; - - let socket_path = match std::env::var_os("HIVE_AGENT_SOCKET") { - Some(p) => std::path::PathBuf::from(p), - None => return axum::Json(serde_json::json!({ "todos": [] })).into_response(), - }; - if !socket_path.exists() { - return axum::Json(serde_json::json!({ "todos": [] })).into_response(); - } - let todos = tokio::time::timeout(std::time::Duration::from_secs(3), async move { - let mut stream = UnixStream::connect(&socket_path).await?; - let req = hive_agent_sock::Request::ListTodos { subsystem: None }; - let mut line = serde_json::to_string(&req)?; - line.push('\n'); - stream.write_all(line.as_bytes()).await?; - stream.flush().await?; - let mut lines = BufReader::new(stream).lines(); - let resp_line = lines - .next_line() - .await? - .ok_or_else(|| anyhow::anyhow!("agent socket closed without response"))?; - let resp: hive_agent_sock::Response = serde_json::from_str(&resp_line)?; - anyhow::Ok(match resp { - hive_agent_sock::Response::LooseEnds { loose_ends } => loose_ends, + let todos = + match crate::todo_server::dial(&hive_agent_sock::Request::ListTodos { subsystem: None }) + .await + { + Some(hive_agent_sock::Response::LooseEnds { loose_ends }) => loose_ends, _ => Vec::new(), - }) - }) - .await - .unwrap_or_else(|_| Err(anyhow::anyhow!("timeout"))) - .unwrap_or_default(); + }; axum::Json(serde_json::json!({ "todos": todos })).into_response() } From dcb2b797154d9b7f261db6bfa8b3925a32910fc7 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 22:38:42 +0200 Subject: [PATCH 8/9] strip issue-tag comments from #2635 inc1 commits per hive-rules --- hive-agent-mcp/src/mcp/mod.rs | 10 +++++----- hive-agent-mcp/src/mcp/render.rs | 4 ++-- hive-agent-sock/src/lib.rs | 6 +++--- hive-agent/src/main.rs | 10 +++++----- hive-agent/src/paths.rs | 6 +++--- hive-agent/src/reminder_timer.rs | 4 ++-- hive-agent/src/reminders.rs | 4 ++-- hive-agent/src/todo_server.rs | 4 ++-- hive-agent/src/web_ui/stats.rs | 4 ++-- 9 files changed, 26 insertions(+), 26 deletions(-) diff --git a/hive-agent-mcp/src/mcp/mod.rs b/hive-agent-mcp/src/mcp/mod.rs index 3a4616c0..177379c7 100644 --- a/hive-agent-mcp/src/mcp/mod.rs +++ b/hive-agent-mcp/src/mcp/mod.rs @@ -344,7 +344,7 @@ impl AgentServer { if is_self_query && let Some(todos) = local_todos().await { loose_ends.extend(todos); } - // Merge local pending reminders (#2635 inc 1 — same self-query-only + // Merge local pending reminders — same self-query-only // restriction: a manager asking for a child's loose-ends no longer // sees the child's reminders, matching the todos precedent above). if is_self_query && let Some(reminders) = local_reminders().await { @@ -453,8 +453,8 @@ impl AgentServer { Err(e) => return e, }; let kind_label = loose_end_kind_label(kind); - // Reminders are harness-local (#2635 inc 1) — dial the in-agent - // socket directly instead of the broker; every other kind + // Reminders are harness-local — dial the in-agent socket + // directly instead of the broker; every other kind // (question/approval) still lives in c0re. if kind == hive_sh4re::CancelLooseEndKind::Reminder { return match dial_agent_socket(&hive_agent_sock::Request::CancelReminder { id }) @@ -543,8 +543,8 @@ impl AgentServer { (Some(s), None) => hive_sh4re::ReminderTiming::InSeconds { seconds: s }, (None, Some(t)) => hive_sh4re::ReminderTiming::At { unix_timestamp: t }, }; - // Reminders are harness-local (#2635 inc 1) — dial the in-agent - // socket directly instead of the broker. + // Reminders are harness-local — dial the in-agent socket + // directly instead of the broker. match dial_agent_socket(&hive_agent_sock::Request::StoreReminder { message: args.message, timing, diff --git a/hive-agent-mcp/src/mcp/render.rs b/hive-agent-mcp/src/mcp/render.rs index 7217b873..fb9821cc 100644 --- a/hive-agent-mcp/src/mcp/render.rs +++ b/hive-agent-mcp/src/mcp/render.rs @@ -338,8 +338,8 @@ pub(super) async fn local_todos() -> Option> { } /// Query the harness's in-agent socket for this agent's local pending -/// reminders (#2635 inc 1 — was a broker query before reminders moved -/// in-container). Same best-effort contract as [`local_todos`]. +/// reminders — was a broker query before reminders moved in-container. +/// Same best-effort contract as [`local_todos`]. pub(super) async fn local_reminders() -> Option> { match dial_agent_socket(&hive_agent_sock::Request::ListReminders).await? { hive_agent_sock::Response::LooseEnds { loose_ends } => Some(loose_ends), diff --git a/hive-agent-sock/src/lib.rs b/hive-agent-sock/src/lib.rs index cf1f503c..a94a5908 100644 --- a/hive-agent-sock/src/lib.rs +++ b/hive-agent-sock/src/lib.rs @@ -1,9 +1,9 @@ //! Wire types for the *in-agent* socket, served by the hive-agent harness //! to the in-container producers (matrix / bash MCP daemons) and //! `forge_notify`. Carries the loose-ends-v2 *todo* op family plus the -//! harness-local *reminder* op family (#2635 increment 1); more in-agent -//! request families may be added over time (the socket is deliberately -//! named for the agent, not the todos). +//! harness-local *reminder* op family; more in-agent request families may +//! be added over time (the socket is deliberately named for the agent, +//! not the todos). //! //! Distinct from `hive-core-agent-sock`, the *host*-served core↔agent //! protocol on `/run/hive/mcp.sock`: this socket never leaves the diff --git a/hive-agent/src/main.rs b/hive-agent/src/main.rs index b18307cb..d4b546a3 100644 --- a/hive-agent/src/main.rs +++ b/hive-agent/src/main.rs @@ -332,8 +332,8 @@ impl Surface for AgentSurface { Ok(Response::LooseEnds { loose_ends }) => u64::try_from(loose_ends.len()).ok(), _ => None, }; - // Reminders are harness-local (#2635 inc 1) — dial the in-agent - // socket directly instead of the broker. + // Reminders are harness-local — dial the in-agent socket directly + // instead of the broker. let reminders = match todo_server::dial(&hive_agent_sock::Request::CountPendingReminders).await { Some(hive_agent_sock::Response::PendingRemindersCount { count }) => Some(count), @@ -458,7 +458,7 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { tracing::error!(error = %e, "web_ui::serve exited with error"); } }); - // Harness-local reminders (#2635 increment 1): a due reminder fires + // Harness-local reminders: a due reminder fires // straight into an mpsc channel the serve loop races against the // broker long-poll (unlike todos, a fire carries real per-row data, // so a bare `Notify` doesn't fit — see `reminder_timer` docs). @@ -476,8 +476,8 @@ async fn serve_main(socket: &Path, poll_ms: u64) -> Result<()> { } }; tokio::spawn(reminder_timer::run(reminder_store.clone(), reminder_tx)); - // In-agent todo socket (loose-ends v2 + #2635 reminders): the harness - // owns the todo + reminder stores locally and serves the + // In-agent todo socket (loose-ends v2 + harness-local reminders): the + // harness owns the todo + reminder stores locally and serves the // in-container producers on `HIVE_AGENT_SOCKET`. A new/changed todo // upsert fires `todo_wake` so the serve loop drives a turn directly — // no broker round-trip, no marker files. Best-effort: if the todos diff --git a/hive-agent/src/paths.rs b/hive-agent/src/paths.rs index ba71a823..9756eb38 100644 --- a/hive-agent/src/paths.rs +++ b/hive-agent/src/paths.rs @@ -48,9 +48,9 @@ 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. +/// Harness-local reminder store. 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") diff --git a/hive-agent/src/reminder_timer.rs b/hive-agent/src/reminder_timer.rs index 53fa1a00..326f8ef5 100644 --- a/hive-agent/src/reminder_timer.rs +++ b/hive-agent/src/reminder_timer.rs @@ -1,5 +1,5 @@ -//! Delivery half of the harness-local reminders store (#2635 increment 1). -//! Polls [`Reminders`] for due rows and pushes each as a +//! Delivery half of the harness-local reminders store. Polls [`Reminders`] +//! for due rows and pushes each as a //! [`hive_sh4re::DeliveredMessage`] down an mpsc channel the serve loop //! races against the broker long-poll — mirrors `todo_server`'s `Notify` //! wake, but a reminder fire carries real per-row data (message/id), so a diff --git a/hive-agent/src/reminders.rs b/hive-agent/src/reminders.rs index a84cd18c..4f4936f3 100644 --- a/hive-agent/src/reminders.rs +++ b/hive-agent/src/reminders.rs @@ -1,6 +1,6 @@ //! 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 +//! in-container reminders migration. 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). //! diff --git a/hive-agent/src/todo_server.rs b/hive-agent/src/todo_server.rs index 5d3a0ccf..ace90cb5 100644 --- a/hive-agent/src/todo_server.rs +++ b/hive-agent/src/todo_server.rs @@ -1,5 +1,5 @@ -//! In-agent socket server (loose-ends v2 + #2635 reminders). Binds the -//! harness-owned `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` +//! In-agent socket server (loose-ends v2 + harness-local reminders). Binds +//! the harness-owned `HIVE_AGENT_SOCKET` and serves the `hive-agent-sock` //! protocol to the in-container producers (matrix / bash daemons, //! forge-notify) and to `hive-agent-mcp`'s `remind`/`get_loose_ends`/ //! `cancel_loose_end` tool impls. Todo ops hit the harness-local [`Todos`] diff --git a/hive-agent/src/web_ui/stats.rs b/hive-agent/src/web_ui/stats.rs index db3271f3..3cbc3ae3 100644 --- a/hive-agent/src/web_ui/stats.rs +++ b/hive-agent/src/web_ui/stats.rs @@ -26,8 +26,8 @@ pub(super) async fn api_stats( } /// Fetch reminder activity stats from the harness-local reminder store over -/// `HIVE_AGENT_SOCKET` (#2635 inc 1 — was a broker RPC before reminders -/// moved in-container). Returns `None` on any transport / decode failure or +/// `HIVE_AGENT_SOCKET` — was a broker RPC before reminders moved +/// in-container. Returns `None` on any transport / decode failure or /// when the socket is unset — the stats are decorative, not authoritative. async fn fetch_reminder_stats(window_secs: u64) -> Option { match crate::todo_server::dial(&hive_agent_sock::Request::ReminderRollup { From a80d0b0fed7c89e8667ea7bb759529fcd0851f87 Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 23:01:16 +0200 Subject: [PATCH 9/9] delete c0re-side reminder plumbing (#2635 inc 1 commit 6) --- hive-c0re/src/container_view.rs | 14 +- hive-c0re/src/coordinator.rs | 20 +- hive-c0re/src/dashboard/mod.rs | 10 - hive-c0re/src/dashboard/reminders.rs | 61 ---- hive-c0re/src/dashboard_events.rs | 13 - hive-c0re/src/loose_ends.rs | 43 +-- hive-c0re/src/main.rs | 8 +- hive-c0re/src/questions.rs | 17 +- hive-c0re/src/server.rs | 14 +- hive-c0re/src/socket_server/mod.rs | 85 ++--- hive-c0re/src/socket_server/reminders.rs | 229 ------------ hive-c0re/src/stats/host_stats.rs | 1 - hive-c0re/src/stores/broker.rs | 365 -------------------- hive-c0re/src/workers/mod.rs | 3 +- hive-c0re/src/workers/reminder_scheduler.rs | 278 --------------- hive-core-agent-sock/src/lib.rs | 45 +-- 16 files changed, 70 insertions(+), 1136 deletions(-) delete mode 100644 hive-c0re/src/dashboard/reminders.rs delete mode 100644 hive-c0re/src/socket_server/reminders.rs delete mode 100644 hive-c0re/src/workers/reminder_scheduler.rs diff --git a/hive-c0re/src/container_view.rs b/hive-c0re/src/container_view.rs index abefef7b..b76d30de 100644 --- a/hive-c0re/src/container_view.rs +++ b/hive-c0re/src/container_view.rs @@ -29,13 +29,6 @@ pub struct ContainerView { /// for this agent's input. #[serde(skip_serializing_if = "Option::is_none")] pub deployed_sha: Option, - /// Count of this agent's pending reminders. Computed during - /// `build_all` via `Broker::count_pending_reminders_for`; the - /// dashboard renders a small chip when > 0. Updates with the - /// 10s `crash_watch` rescan + every container mutation site; - /// not real-time on remind/cancel-reminder but close enough. - #[serde(default)] - pub pending_reminders: u64, /// Name of this agent's parent in the agent hierarchy. `None` /// marks the agent as root-level; the dashboard renders it without /// indentation. Sourced from `meta/topology.json` (single source of @@ -55,7 +48,7 @@ pub struct ContainerView { /// Build the full container list. Wraps `lifecycle::list()` and /// resolves every per-agent attribute the dashboard surfaces. -pub async fn build_all(coord: &Coordinator) -> Vec { +pub async fn build_all() -> Vec { let raw = lifecycle::list().await.unwrap_or_default(); let locked = read_meta_locked_revs(); // Pull the topology map once and look up each agent's parent below. @@ -79,10 +72,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec { let needs_update = crate::auto_update::agent_config_pending(logical.as_str(), deployed_full).await; let deployed_sha = deployed_full.map(|s| s[..s.len().min(12)].to_owned()); - let pending_reminders = coord - .broker - .count_pending_reminders_for(logical.as_str()) - .unwrap_or(0); let parent = topology.get(logical.as_str()).cloned().flatten(); let running = lifecycle::is_running(logical.as_str()).await; // needs_login fires when EITHER the claude session dir is missing @@ -108,7 +97,6 @@ pub async fn build_all(coord: &Coordinator) -> Vec { needs_update, needs_login, deployed_sha, - pending_reminders, parent, active_model, }); diff --git a/hive-c0re/src/coordinator.rs b/hive-c0re/src/coordinator.rs index 6c07db17..c24247f9 100644 --- a/hive-c0re/src/coordinator.rs +++ b/hive-c0re/src/coordinator.rs @@ -612,24 +612,6 @@ impl Coordinator { .map(|m| m.keys().cloned().collect()) } - /// Emit a `RemindersChanged` snapshot event. Called from every - /// reminder mutation site (agent `remind` calls, operator cancel / - /// retry, and the scheduler after each delivery batch) so the - /// dashboard's pending-reminders list stays live without polling. - pub fn emit_reminders_snapshot(self: &Arc) { - let reminders = match self.broker.list_pending_reminders() { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "emit_reminders_snapshot: list failed"); - return; - } - }; - self.emit_dashboard_event(DashboardEvent::RemindersChanged { - seq: self.next_seq(), - reminders, - }); - } - /// Emit a `CapabilitiesChanged` snapshot event. Called from the /// rebuild-queue worker after a `PermChange` / Capabilities entry /// commits the JSON file, so the P3RM1SS10NS tab updates live. @@ -930,7 +912,7 @@ impl Coordinator { /// Cheap when nothing changed (one `nixos-container list` + a /// `HashMap` diff + zero emits). pub async fn rescan_containers_and_emit(self: &Arc) { - let fresh = container_view::build_all(self).await; + let fresh = container_view::build_all().await; let mut last = self.last_containers.lock().await; let mut changed_or_new = Vec::new(); let mut removed = Vec::new(); diff --git a/hive-c0re/src/dashboard/mod.rs b/hive-c0re/src/dashboard/mod.rs index 00a65b4c..580feeba 100644 --- a/hive-c0re/src/dashboard/mod.rs +++ b/hive-c0re/src/dashboard/mod.rs @@ -32,7 +32,6 @@ mod meta_inputs; mod misc_api; pub(crate) mod permissions; mod questions; -mod reminders; mod schedules; mod state_files; mod state_snapshot; @@ -93,7 +92,6 @@ pub async fn serve( "/api/extra-forge-account", post(extra_forges::post_extra_forge_account), ) - .route("/api/reminders", get(reminders::api_reminders)) .route("/api/operator-inbox", get(misc_api::api_operator_inbox)) .route("/api/stats-hive", get(misc_api::api_stats_hive)) .route( @@ -212,14 +210,6 @@ pub async fn serve( "/api/github-account", post(matrix_accounts::post_github_account).get(matrix_accounts::get_github_account), ) - .route( - "/api/cancel-reminder/{id}", - post(reminders::post_cancel_reminder), - ) - .route( - "/api/retry-reminder/{id}", - post(reminders::post_retry_reminder), - ) .route("/api/request-spawn", post(misc_api::post_request_spawn)) .route("/api/op-send", post(misc_api::post_op_send)) .route("/api/meta-update", post(meta_inputs::post_meta_update)) diff --git a/hive-c0re/src/dashboard/reminders.rs b/hive-c0re/src/dashboard/reminders.rs deleted file mode 100644 index d61550b8..00000000 --- a/hive-c0re/src/dashboard/reminders.rs +++ /dev/null @@ -1,61 +0,0 @@ -//! Reminder endpoints for the dashboard. -//! -//! Lists pending reminders for the reminders tab, and lets the operator -//! cancel a pending reminder or reset its failure state so the scheduler -//! retries it on the next tick. - -use axum::{ - extract::{Path as AxumPath, State}, - http::StatusCode, - response::{IntoResponse, Response}, -}; - -use problem_details::ProblemDetails; - -use super::{AppState, error_problem, error_response}; - -pub(super) async fn api_reminders(State(state): State) -> Response { - match state.coord.broker.list_pending_reminders() { - Ok(rows) => axum::Json(rows).into_response(), - Err(e) => error_response(&format!("reminders: {e:#}")), - } -} - -pub(super) async fn post_cancel_reminder( - State(state): State, - AxumPath(id): AxumPath, -) -> Result { - match state.coord.broker.cancel_reminder(id) { - Ok(0) => Err(ProblemDetails::from_status_code(StatusCode::NOT_FOUND) - .with_detail(format!("reminder {id} not pending (already delivered?)"))), - Ok(_) => { - tracing::info!(%id, "operator cancelled reminder"); - state.coord.emit_reminders_snapshot(); - Ok((StatusCode::OK, "ok").into_response()) - } - Err(e) => Err(error_problem(&format!( - "cancel reminder {id} failed: {e:#}" - ))), - } -} - -/// Reset a pending reminder's failure state so the scheduler -/// retries it on the next tick. Useful when the failure was -/// transient (sqlite lock contention, disk full → freed up) and -/// the operator wants delivery to resume immediately instead of -/// the row sitting in attempt-count-capped purgatory. -pub(super) async fn post_retry_reminder( - State(state): State, - AxumPath(id): AxumPath, -) -> Result { - match state.coord.broker.reset_reminder_failure(id) { - Ok(0) => Err(ProblemDetails::from_status_code(StatusCode::NOT_FOUND) - .with_detail(format!("reminder {id} not pending (already delivered?)"))), - Ok(_) => { - tracing::info!(%id, "operator reset reminder failure for retry"); - state.coord.emit_reminders_snapshot(); - Ok((StatusCode::OK, "ok").into_response()) - } - Err(e) => Err(error_problem(&format!("retry reminder {id} failed: {e:#}"))), - } -} diff --git a/hive-c0re/src/dashboard_events.rs b/hive-c0re/src/dashboard_events.rs index d3fb9c92..b5451b05 100644 --- a/hive-c0re/src/dashboard_events.rs +++ b/hive-c0re/src/dashboard_events.rs @@ -219,14 +219,6 @@ pub enum DashboardEvent { seq: u64, schedules: Vec, }, - /// Full snapshot of all pending reminders. Emitted after every - /// reminder mutation: agent `remind` calls, operator cancel / retry, - /// and the scheduler tick after each delivery batch. Lets the - /// dashboard's reminders section stay live without polling. - RemindersChanged { - seq: u64, - reminders: Vec, - }, /// Full snapshot of capability grants (per-agent `Vec`). /// Emitted from the rebuild-queue worker after a `PermChange` /// `Capabilities` entry commits the JSON file. Lets the P3RM1SS10NS @@ -294,7 +286,6 @@ impl DashboardEvent { DashboardEvent::MetaUpdateRunning { .. } => "meta_update_running", DashboardEvent::RebuildQueueChanged { .. } => "rebuild_queue_changed", DashboardEvent::SchedulesChanged { .. } => "schedules_changed", - DashboardEvent::RemindersChanged { .. } => "reminders_changed", DashboardEvent::CapabilitiesChanged { .. } => "capabilities_changed", DashboardEvent::ToolGroupsChanged { .. } => "tool_groups_changed", DashboardEvent::AuditEntryAdded { .. } => "audit_entry_added", @@ -420,10 +411,6 @@ mod tests { seq: 1, schedules: Vec::new(), }, - DashboardEvent::RemindersChanged { - seq: 1, - reminders: Vec::new(), - }, DashboardEvent::CapabilitiesChanged { seq: 1, caps: Vec::new(), diff --git a/hive-c0re/src/loose_ends.rs b/hive-c0re/src/loose_ends.rs index d510d222..c206b19f 100644 --- a/hive-c0re/src/loose_ends.rs +++ b/hive-c0re/src/loose_ends.rs @@ -3,7 +3,8 @@ //! a single agent (`for_agent`) or the whole hive (`hive_wide`). //! `Request::GetLooseEnds` from either the agent or manager socket //! lands here so the routing logic + age-seconds derivation stay in -//! one place. +//! one place. Reminders are agent-local (in-container store) and no +//! longer sourced from here. //! //! Call frequency is low (an agent doing self-introspection between //! turns), so the sweep happens fresh every time — no caching, no @@ -27,18 +28,17 @@ use hive_sh4re::wire_time::now_unix; /// root submits for top-level agents). Legacy rows with no recorded /// submitter count as the root's; /// - unanswered questions where `agent` is the asker (waiting on -/// someone) OR the target (owes a reply); -/// - pending reminders this agent scheduled (`owner == self`). +/// someone) OR the target (owes a reply). /// -/// Ordered `pending_messages` (when non-zero) → approvals → questions → -/// reminders within the returned vector. Within each kind, -/// source-of-truth ordering (sqlite's `pending()` queries return -/// newest-first within their indexes). +/// Ordered `pending_messages` (when non-zero) → approvals → questions +/// within the returned vector. Within each kind, source-of-truth +/// ordering (sqlite's `pending()` queries return newest-first within +/// their indexes). /// /// # Errors /// /// Propagates errors from `count_pending` and the pending-approval / -/// question / reminder sqlite queries. +/// question sqlite queries. pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -85,25 +85,13 @@ pub fn for_agent(coord: &Coordinator, agent: &str) -> Result> { age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } - for r in coord.broker.list_pending_reminders()? { - if r.agent != agent { - continue; - } - out.push(LooseEnd::Reminder { - id: r.id, - owner: r.agent, - message: r.message, - due_at: r.due_at, - age_seconds: saturating_age(now, r.created_at.timestamp()), - }); - } Ok(out) } /// Hive-wide loose-ends view: EVERY pending approval + EVERY -/// unanswered question + EVERY pending reminder. Manager surface -/// only; sub-agents can't see each other's threads via the agent -/// surface (`for_agent` filters by name). +/// unanswered question. Manager surface only; sub-agents can't see +/// each other's threads via the agent surface (`for_agent` filters by +/// name). pub fn hive_wide(coord: &Coordinator) -> Result> { let now = now_unix(); let mut out = Vec::new(); @@ -125,15 +113,6 @@ pub fn hive_wide(coord: &Coordinator) -> Result> { age_seconds: saturating_age(now, q.asked_at.timestamp()), }); } - for r in coord.broker.list_pending_reminders()? { - out.push(LooseEnd::Reminder { - id: r.id, - owner: r.agent, - message: r.message, - due_at: r.due_at, - age_seconds: saturating_age(now, r.created_at.timestamp()), - }); - } Ok(out) } diff --git a/hive-c0re/src/main.rs b/hive-c0re/src/main.rs index a97aff7b..f8e2870d 100644 --- a/hive-c0re/src/main.rs +++ b/hive-c0re/src/main.rs @@ -45,8 +45,7 @@ pub(crate) use stores::{ approvals, audit_log, broker, build_logs, db, operator_questions, power, scheduled_prompts, }; pub(crate) use workers::{ - agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, reminder_scheduler, - scheduled_prompts_worker, + agent_sockets, auto_update, crash_watch, knowledge, mcp_sockets, scheduled_prompts_worker, }; use coordinator::{Coordinator, HiveEnv, ServeConfig}; @@ -214,7 +213,7 @@ fn matrix_sweep_banner(ctx: sweep_health::SweepFailure) -> String { } /// Start the coordinator daemon: open the broker, run migrations, spawn -/// background tasks (auto-update, vacuums, crash-watcher, reminder-scheduler, +/// background tasks (auto-update, vacuums, crash-watcher, scheduled-prompts, /// dashboard), then serve the admin socket until a signal arrives. #[allow( clippy::too_many_lines, @@ -522,9 +521,6 @@ async fn cmd_serve( // run_reconcile call register_agent on start; kill/destroy call // unregister_agent. No recurring poll needed — c0re owns the listeners. mcp_sockets::sync_on_start(coord.clone()).await; - // Reminder scheduler: drains due reminders + handles - // file_path payload persistence. See reminder_scheduler.rs. - reminder_scheduler::spawn(coord.clone()); // Scheduled-prompts worker: drains due scheduled_prompts rows // and fans the body out to each active target's inbox. See // scheduled_prompts_worker.rs. diff --git a/hive-c0re/src/questions.rs b/hive-c0re/src/questions.rs index d531ee0c..391d6451 100644 --- a/hive-c0re/src/questions.rs +++ b/hive-c0re/src/questions.rs @@ -186,14 +186,15 @@ pub fn handle_cancel_loose_end( Ok(()) } hive_sh4re::CancelLooseEndKind::Reminder => { - // Agent-socket path: ownership-only (cancel your own reminder). - let owner = coord - .broker - .cancel_reminder_as(id, canceller, false) - .map_err(|e| format!("{e:#}"))?; - tracing::info!(%id, %canceller, %owner, "reminder cancelled"); - coord.emit_reminders_snapshot(); - Ok(()) + // Reminders are now agent-local (in-container store) — the + // agent-mcp `cancel_loose_end` tool branches on this kind and + // dials the agent's own socket directly, never forwarding to + // hive-c0re. This arm should be unreachable in practice; kept + // only so the match stays exhaustive. + Err(format!( + "reminder {id}: reminders are handled locally by the agent, \ + not by hive-c0re" + )) } hive_sh4re::CancelLooseEndKind::Approval => { // Withdrawing an approval needs the grantable `approvals` diff --git a/hive-c0re/src/server.rs b/hive-c0re/src/server.rs index 21089867..4bfb3bf4 100644 --- a/hive-c0re/src/server.rs +++ b/hive-c0re/src/server.rs @@ -159,7 +159,7 @@ async fn dispatch(req: &HostRequest, coord: Arc) -> HostResponse { HostResponse::dags(dags) } HostRequest::List => HostResponse::list(lifecycle::list().await?), - HostRequest::AgentStatus => handle_agent_status(&coord).await, + HostRequest::AgentStatus => handle_agent_status().await, // The hive domain + per-surface public URLs are injected into // c0re's service env by hive-c0re.nix; surface them so the // operator CLI can fill in this hive's own identity (the @@ -298,8 +298,8 @@ async fn handle_spawn(coord: &Arc, name: &str) -> Result) -> HostResponse { - let rows = crate::container_view::build_all(coord) +async fn handle_agent_status() -> HostResponse { + let rows = crate::container_view::build_all() .await .into_iter() .map(|v| hive_sh4re::AgentStatusRow { @@ -308,7 +308,13 @@ async fn handle_agent_status(coord: &Arc) -> HostResponse { needs_update: v.needs_update, needs_login: v.needs_login, deployed_sha: v.deployed_sha, - pending_reminders: v.pending_reminders, + // Reminders are agent-local now; c0re has no cross-agent + // visibility into pending counts anymore. Stubbed + // to 0 rather than deleting the wire field outright — leaves + // `hivectl status`/the dashboard column intact syntactically, + // just always empty, until iris's frontend follow-up decides + // whether to drop the column entirely. + pending_reminders: 0, parent: v.parent, }) .collect(); diff --git a/hive-c0re/src/socket_server/mod.rs b/hive-c0re/src/socket_server/mod.rs index ffe3c81f..e9cfc3d0 100644 --- a/hive-c0re/src/socket_server/mod.rs +++ b/hive-c0re/src/socket_server/mod.rs @@ -23,7 +23,6 @@ use crate::coordinator::Coordinator; mod config_approvals; mod lifecycle_handlers; -mod reminders; mod schedules; pub(crate) use config_approvals::submit_merge_config_pr; @@ -34,7 +33,6 @@ use config_approvals::{handle_request_init_config, handle_request_update_meta_in use lifecycle_handlers::{ handle_kill, handle_list_descendants, handle_restart, handle_start, handle_update, }; -use reminders::{handle_remind, resolve_agent_state_target}; use schedules::{ EditSchedulePatch, handle_cancel_schedule, handle_edit_schedule, handle_fire_schedule_now, handle_list_schedules, handle_request_schedule_prompt, @@ -184,8 +182,8 @@ pub(crate) fn recv_timeout(wait_seconds: Option) -> std::time::Duration { /// Handle the subset of `Request` variants that are identical on both /// the agent socket and the manager socket. Returns `Some(response)` for /// every variant it handles; returns `None` for variants with socket-specific -/// semantics (e.g. `GetLooseEnds` / `CountPendingReminders` / `ReminderRollup` -/// where the manager can target other agents) or for manager-only variants. +/// semantics (e.g. `GetLooseEnds`, where the manager can target other +/// agents) or for manager-only variants. /// /// The unified `dispatch` calls this first; the remaining arms (which gate /// on topology / capabilities / tool-groups) are handled there. @@ -234,11 +232,6 @@ pub(crate) async fn dispatch_shared( |()| hive_core_agent_sock::Response::Ok, ) } - hive_core_agent_sock::Request::Remind { - message, - timing, - file_path, - } => handle_remind(coord, agent, message, timing, file_path.as_deref()), hive_core_agent_sock::Request::SetStatus { text } => handle_set_status(coord, text), hive_core_agent_sock::Request::GetAgentMeta { name } => { handle_get_agent_meta(coord, agent, name.as_ref()).await @@ -590,13 +583,6 @@ async fn dispatch(req: &Request, agent: &str, coord: &Arc) -> Respo Request::GetLooseEnds { agent: target } => { handle_get_loose_ends(coord, agent, target.as_deref()) } - Request::CountPendingReminders { agent: target } => { - handle_count_pending_reminders(coord, agent, target.as_deref()) - } - Request::ReminderRollup { - since_secs, - agent: target, - } => handle_reminder_rollup(coord, agent, target.as_deref(), *since_secs), // Orchestration / diagnostics verbs — gated per-verb on tool-group // membership or topology (see `dispatch_orchestration`). _ => dispatch_orchestration(req, agent, coord).await, @@ -791,41 +777,38 @@ fn handle_get_loose_ends(coord: &Arc, agent: &str, target: Option<& } } -/// `CountPendingReminders` — resolve the target (own / subtree free, else -/// `QueryAgentState`) then count its pending reminders. -fn handle_count_pending_reminders( - coord: &Arc, - agent: &str, - target: Option<&str>, -) -> Response { - match resolve_agent_state_target(agent, target) { - Ok(name) => match coord.broker.count_pending_reminders_for(name) { - Ok(count) => Response::PendingRemindersCount { count }, - Err(e) => Response::Err { - message: format!("{e:#}"), - }, - }, - Err(message) => Response::Err { message }, - } -} - -/// `ReminderRollup` — resolve the target (own / subtree free, else -/// `QueryAgentState`) then roll up its reminders fired in the last -/// `since_secs`. -fn handle_reminder_rollup( - coord: &Arc, - agent: &str, - target: Option<&str>, - since_secs: u64, -) -> Response { - match resolve_agent_state_target(agent, target) { - Ok(name) => match coord.broker.reminder_rollup_for(name, since_secs) { - Ok(stats) => Response::ReminderRollup(stats), - Err(e) => Response::Err { - message: format!("{e:#}"), - }, - }, - Err(message) => Response::Err { message }, +/// Resolve the target agent name for a *named* `GetLooseEnds` query. Rules: +/// +/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). +/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. +/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. +/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. +fn resolve_agent_state_target<'a>( + caller: &'a str, + target: Option<&'a str>, +) -> Result<&'a str, String> { + match target { + None => Ok(caller), + Some("*") => Err( + "hive-wide query (agent=\"*\") is only valid for loose-ends; \ + not available for this query" + .to_owned(), + ), + Some(name) => { + // Own subtree (the root covers all) is visible without extra + // capability; `is_descendant_of` returns true for `name == caller`. + if crate::topology::is_descendant_of(name, caller) { + return Ok(name); + } + if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { + Ok(name) + } else { + Err(format!( + "agent `{caller}` cannot query `{name}`: not in its subtree and \ + `query_agent_state` capability is not granted" + )) + } + } } } diff --git a/hive-c0re/src/socket_server/reminders.rs b/hive-c0re/src/socket_server/reminders.rs deleted file mode 100644 index ae879da1..00000000 --- a/hive-c0re/src/socket_server/reminders.rs +++ /dev/null @@ -1,229 +0,0 @@ -//! Reminder request handling: the `Remind` handler, the shared -//! `store_remind` storage path with its pending-cap and large-body -//! auto-save dance, timing resolution, and the agent-state target -//! resolution shared by the loose-ends / reminder query handlers. - -use std::sync::Arc; - -use hive_core_agent_sock::Response; - -use crate::coordinator::Coordinator; - -pub(super) fn handle_remind( - coord: &Arc, - agent: &str, - message: &str, - timing: &hive_sh4re::ReminderTiming, - file_path: Option<&str>, -) -> Response { - match store_remind(coord, agent, message, timing, file_path) { - Ok(()) => Response::Ok, - Err(message) => Response::Err { message }, - } -} - -/// Shared remind-storage path used by both the agent and the manager -/// dispatchers. Validates timing, applies the auto-file overflow -/// dance (see [`prepare_remind_storage`]), and writes the reminder -/// row. Returns `Ok(())` on success, or a caller-ready error string -/// the dispatcher wraps in `*Response::Err`. -/// Maximum pending (un-delivered) reminders per agent. Exceeding this -/// causes `store_remind` to return an error so the agent knows to back -/// off instead of silently dropping. Override via -/// `HIVE_REMIND_MAX_PENDING_PER_AGENT`; set to `0` to disable the cap -/// (not recommended — a runaway agent can still flood the scheduler). -const DEFAULT_REMIND_MAX_PENDING: u64 = 50; - -fn remind_max_pending() -> u64 { - std::env::var("HIVE_REMIND_MAX_PENDING_PER_AGENT") - .ok() - .and_then(|s| s.trim().parse::().ok()) - .unwrap_or(DEFAULT_REMIND_MAX_PENDING) -} - -pub(crate) fn store_remind( - coord: &Arc, - agent: &str, - message: &str, - timing: &hive_sh4re::ReminderTiming, - file_path: Option<&str>, -) -> Result<(), String> { - let max = remind_max_pending(); - if max > 0 { - let pending = coord.broker.count_pending_reminders_for(agent).unwrap_or(0); - if pending >= max { - return Err(format!( - "reminder rejected: agent `{agent}` already has {pending} pending \ - reminders (cap {max}). Cancel some via `cancel_loose_end` or wait \ - for them to fire before scheduling more. Override the cap with \ - `HIVE_REMIND_MAX_PENDING_PER_AGENT`." - )); - } - } - let due_at = resolve_due_at(timing).map_err(|e| format!("invalid reminder timing: {e:#}"))?; - let (stored_message, stored_path) = prepare_remind_storage(agent, message, file_path)?; - let id = coord - .broker - .store_reminder(agent, &stored_message, stored_path.as_deref(), due_at) - .map_err(|e| format!("failed to store reminder: {e:#}"))?; - tracing::info!(%id, %agent, %due_at, "reminder scheduled"); - coord.emit_reminders_snapshot(); - Ok(()) -} - -/// Decide what we actually store in the reminders row, applying the -/// same byte cap as the rest of the wire protocol -/// ([`crate::limits::MESSAGE_MAX_BYTES`]). Three outcomes: -/// -/// 1. Body within the cap → stored verbatim, with whatever `file_path` -/// the caller passed (None or Some). The scheduler honours -/// `file_path` at delivery time as before. -/// 2. Body over the cap, no caller `file_path` → auto-generate a path -/// under `/agents//state/reminders/auto-.md`, write the -/// body to disk now, store a short pointer hint as the message and -/// clear `file_path` (so the scheduler doesn't re-write at -/// delivery and overwrite the body with the hint). -/// 3. Body over the cap, caller provided `file_path` → honour the -/// caller's path: write the body to it now, store the same hint -/// and clear `file_path` for the same reason as (2). -/// -/// Returns `(stored_message, stored_file_path)` on success, or a -/// caller-ready error string on auto-save failure (which is the only -/// way a Remind request can be refused for size — the agent never has -/// to think about the cap). -fn prepare_remind_storage( - agent: &str, - message: &str, - file_path: Option<&str>, -) -> Result<(String, Option), String> { - if message.len() <= crate::limits::MESSAGE_MAX_BYTES { - return Ok((message.to_owned(), file_path.map(str::to_owned))); - } - let req_path = match file_path { - Some(p) => p.to_owned(), - None => auto_reminder_path(agent), - }; - let host_path = crate::reminder_scheduler::resolve_host_path(agent, &req_path) - .map_err(|reason| format!("auto-save path `{req_path}` rejected: {reason}"))?; - crate::reminder_scheduler::write_payload(agent, &host_path, message).map_err(|reason| { - format!("auto-save of large reminder body to `{req_path}` failed: {reason}") - })?; - let hint = format!( - "[reminder body of {} bytes auto-saved to `{req_path}`; read with your filesystem tools]", - message.len() - ); - Ok((hint, None)) -} - -/// Generate a per-agent path for an auto-saved reminder body. Uses -/// `unix_nanos` plus the agent name to keep collisions infinitesimal -/// across the agent's own state subtree (we're not stamping a hostname -/// since hive-c0re is single-host). -fn auto_reminder_path(agent: &str) -> String { - let ts_ns = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .map_or(0, |d| d.as_nanos()); - format!("/agents/{agent}/state/reminders/auto-{ts_ns}.md") -} - -/// Resolve the target agent name for a *named* `GetLooseEnds` / -/// `CountPendingReminders` / `ReminderRollup` query. Rules: -/// -/// - `None` (or `Some(caller)`) → the caller's own threads (always allowed). -/// - `Some(descendant)` in the caller's subtree (the root's subtree is the whole hive) → allowed, no extra capability. -/// - `Some(other)` outside the subtree → requires the `query_agent_state` capability; error otherwise. -/// - `Some("*")` → rejected here; the hive-wide sweep is handled by `handle_get_loose_ends` under the same `query_agent_state` gate. -pub(super) fn resolve_agent_state_target<'a>( - caller: &'a str, - target: Option<&'a str>, -) -> Result<&'a str, String> { - match target { - None => Ok(caller), - Some("*") => Err( - "hive-wide query (agent=\"*\") is only valid for loose-ends; \ - not available for this query" - .to_owned(), - ), - Some(name) => { - // Own subtree (the root covers all) is visible without extra - // capability; `is_descendant_of` returns true for `name == caller`. - if crate::topology::is_descendant_of(name, caller) { - return Ok(name); - } - if crate::capabilities::has_cap(caller, hive_sh4re::Capability::QueryAgentState) { - Ok(name) - } else { - Err(format!( - "agent `{caller}` cannot query `{name}`: not in its subtree and \ - `query_agent_state` capability is not granted" - )) - } - } - } -} - -/// Resolve the `due_at` unix timestamp for a Remind request. Returns -/// distinct error messages for each failure mode (overflow on -/// `InSeconds`, pre-epoch clock, `i64` cast wrap) so the caller can tell -/// what went wrong without inspecting the chain. -fn resolve_due_at(timing: &hive_sh4re::ReminderTiming) -> anyhow::Result { - use hive_sh4re::ReminderTiming; - match timing { - ReminderTiming::InSeconds { seconds } => { - let now = std::time::SystemTime::now(); - let future = now - .checked_add(std::time::Duration::from_secs(*seconds)) - .ok_or_else(|| { - anyhow::anyhow!("InSeconds overflow: {seconds}s exceeds system time range") - })?; - let duration = future - .duration_since(std::time::UNIX_EPOCH) - .map_err(|e| anyhow::anyhow!("system time before UNIX_EPOCH: {e}"))?; - i64::try_from(duration.as_secs()) - .map_err(|e| anyhow::anyhow!("unix timestamp exceeds i64 range: {e}")) - } - ReminderTiming::At { unix_timestamp } => Ok(*unix_timestamp), - } -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn auto_reminder_path_format() { - let p = auto_reminder_path("damocles"); - assert!(p.starts_with("/agents/damocles/state/reminders/auto-")); - assert!( - std::path::Path::new(&p) - .extension() - .is_some_and(|ext| ext.eq_ignore_ascii_case("md")) - ); - } - - #[test] - fn prepare_remind_storage_passthrough_under_cap() { - let (msg, fp) = prepare_remind_storage("foo", "small body", None).unwrap(); - assert_eq!(msg, "small body"); - assert_eq!(fp, None); - } - - #[test] - fn prepare_remind_storage_passthrough_with_caller_file_path() { - let (msg, fp) = - prepare_remind_storage("foo", "small", Some("/agents/foo/state/x.md")).unwrap(); - assert_eq!(msg, "small"); - assert_eq!(fp.as_deref(), Some("/agents/foo/state/x.md")); - } - - #[test] - fn resolve_agent_state_target_self_and_default_are_free() { - // No topology/capability state needed for these: `None` and the - // caller's own name resolve to the caller (`is_descendant_of` short- - // circuits to true when candidate == ancestor); `"*"` is rejected - // (the hive-wide sweep is handled by the loose-ends caller instead). - assert_eq!(resolve_agent_state_target("iris", None), Ok("iris")); - assert_eq!(resolve_agent_state_target("iris", Some("iris")), Ok("iris")); - assert!(resolve_agent_state_target("iris", Some("*")).is_err()); - } -} diff --git a/hive-c0re/src/stats/host_stats.rs b/hive-c0re/src/stats/host_stats.rs index 0e8bcaf5..1ccf80bd 100644 --- a/hive-c0re/src/stats/host_stats.rs +++ b/hive-c0re/src/stats/host_stats.rs @@ -250,7 +250,6 @@ mod tests { needs_update: false, needs_login, deployed_sha: None, - pending_reminders: 0, parent: None, active_model: None, } diff --git a/hive-c0re/src/stores/broker.rs b/hive-c0re/src/stores/broker.rs index 1b8cca58..729bbeed 100644 --- a/hive-c0re/src/stores/broker.rs +++ b/hive-c0re/src/stores/broker.rs @@ -7,7 +7,6 @@ use std::sync::Mutex; use anyhow::{Context, Result}; -use chrono::{DateTime, Utc}; use hive_sh4re::wire_time::now_unix; use hive_sh4re::{InboxRow, Message}; @@ -30,18 +29,6 @@ CREATE TABLE IF NOT EXISTS messages ( CREATE INDEX IF NOT EXISTS idx_messages_undelivered ON messages (recipient, priority DESC, id) WHERE delivered_at IS NULL; -CREATE TABLE IF NOT EXISTS reminders ( - id INTEGER PRIMARY KEY AUTOINCREMENT, - agent TEXT NOT NULL, - 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 (agent, due_at) WHERE sent_at IS NULL; - CREATE TABLE IF NOT EXISTS kv ( key TEXT PRIMARY KEY, value TEXT NOT NULL @@ -52,12 +39,6 @@ CREATE TABLE IF NOT EXISTS kv ( /// may drop events past this; we send a `lagged` notice in their stream. const EVENT_CHANNEL: usize = 256; -/// Row shape returned by [`Broker::get_due_reminders`]: -/// `(agent, reminder_id, message, file_path)`. Type alias keeps -/// `clippy::type_complexity` quiet and makes the scheduler call site -/// self-documenting. -pub type DueReminder = (String, i64, String, Option); - /// A single message hand-off from broker to recipient. Carries the /// broker's row id (so the harness can drive `ack_turn` later) and /// the redelivery flag (so the harness can prepend the @@ -70,37 +51,6 @@ pub struct Delivery { pub message: Message, } -/// Row shape for [`Broker::list_pending_reminders`], shipped on the -/// dashboard `/api/reminders` response. -#[derive(Debug, Clone, Serialize)] -pub struct PendingReminder { - pub id: i64, - pub agent: String, - pub message: String, - #[serde(skip_serializing_if = "Option::is_none")] - pub file_path: Option, - pub due_at: DateTime, - pub created_at: DateTime, - /// Most recent delivery failure for this row, if any. Cleared - /// to NULL on operator retry. Surfaced inline in the dashboard - /// so a stuck reminder doesn't just silently retry forever. - #[serde(skip_serializing_if = "Option::is_none")] - pub last_error: Option, - /// Number of failed delivery attempts since the row was - /// created or last retried. After `MAX_REMINDER_ATTEMPTS` the - /// scheduler stops trying (the row stays in `pending` with the - /// error so the operator can decide between retry + cancel). - #[serde(default)] - pub attempt_count: u32, -} - -/// Stop retrying a row after this many consecutive failures. The -/// scheduler quits scheduling it until an operator explicitly -/// retries (which resets the counter) or cancels (which deletes -/// the row). Below the cap the existing 5s tick re-attempts each -/// time the row is due. -pub const MAX_REMINDER_ATTEMPTS: u32 = 5; - /// Intra-process broker event. `recv_blocking_batch` listens on the /// same channel as the dashboard forwarder; the forwarder re-emits /// each event as a `DashboardEvent` with a freshly-stamped seq from @@ -193,18 +143,6 @@ const BROKER_MIGRATIONS: &[Migration] = &[ COMMIT;", adds_column: Some(("messages", "priority")), }, - // v4: attempt_count on reminders for the MAX_REMINDER_ATTEMPTS cap. - Migration { - sql: "ALTER TABLE reminders ADD COLUMN \ - attempt_count INTEGER NOT NULL DEFAULT 0", - adds_column: Some(("reminders", "attempt_count")), - }, - // v5: last_error on reminders — last delivery failure surfaced on the - // dashboard so a stuck reminder is visible without digging in logs. - Migration { - sql: "ALTER TABLE reminders ADD COLUMN last_error TEXT", - adds_column: Some(("reminders", "last_error")), - }, ]; impl Broker { @@ -850,309 +788,6 @@ impl Broker { } Ok(u64::try_from(n).unwrap_or(0)) } - - /// Store a new reminder. Returns the reminder id. - pub fn store_reminder( - &self, - agent: &str, - message: &str, - file_path: Option<&str>, - due_at: i64, - ) -> Result { - let conn = self.conn.lock().unwrap(); - conn.execute( - "INSERT INTO reminders (agent, message, file_path, due_at, created_at) VALUES (?1, ?2, ?3, ?4, ?5)", - params![agent, message, file_path, due_at, now_unix()], - )?; - let id = conn.last_insert_rowid(); - Ok(id) - } - - /// Every reminder still pending delivery, newest-first. Used by the - /// dashboard's reminders pane so the operator can see what's queued - /// + cancel rows that are no longer wanted. - pub fn list_pending_reminders(&self) -> Result> { - let conn = self.conn.lock().unwrap(); - let mut stmt = conn.prepare( - "SELECT id, agent, message, file_path, due_at, created_at, \ - last_error, attempt_count \ - FROM reminders \ - WHERE sent_at IS NULL \ - ORDER BY due_at ASC", - )?; - let rows = stmt.query_map([], |row| { - let attempts: i64 = row.get(7)?; - Ok(PendingReminder { - id: row.get(0)?, - agent: row.get(1)?, - message: row.get(2)?, - file_path: row.get(3)?, - due_at: hive_sh4re::wire_time::from_secs(row.get(4)?), - created_at: hive_sh4re::wire_time::from_secs(row.get(5)?), - last_error: row.get(6)?, - attempt_count: u32::try_from(attempts).unwrap_or(0), - }) - })?; - rows.collect::>>() - .context("list pending reminders") - } - - /// Mark a delivery attempt as failed: bump `attempt_count` and - /// stash the error string. Called by `reminder_scheduler::tick` - /// when `deliver_reminder` returns Err. Soft-cap behaviour - /// lives in `get_due_reminders` (rows over the cap drop out - /// of the due-list and stop being attempted until retry). - pub fn record_reminder_failure(&self, id: i64, reason: &str) -> Result<()> { - let conn = self.conn.lock().unwrap(); - conn.execute( - "UPDATE reminders \ - SET attempt_count = attempt_count + 1, last_error = ?1 \ - WHERE id = ?2 AND sent_at IS NULL", - params![reason, id], - )?; - Ok(()) - } - - /// Clear the failure state on a pending reminder so the - /// scheduler picks it up again. No-op when the row is already - /// fresh (`attempt_count == 0`). Returns the number of rows - /// affected so callers can distinguish "retried" from "no - /// such pending reminder" (already delivered, or wrong id). - pub fn reset_reminder_failure(&self, id: i64) -> Result { - let conn = self.conn.lock().unwrap(); - let n = conn.execute( - "UPDATE reminders \ - SET attempt_count = 0, last_error = NULL \ - WHERE id = ?1 AND sent_at IS NULL", - params![id], - )?; - Ok(n) - } - - /// Count this agent's still-pending (un-delivered) reminders. - /// Used by the per-turn stats sink for a cheap "what was queued - /// at turn-end" snapshot. - pub fn count_pending_reminders_for(&self, agent: &str) -> Result { - let conn = self.conn.lock().unwrap(); - let n: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND sent_at IS NULL", - params![agent], - |row| row.get(0), - )?; - Ok(u64::try_from(n).unwrap_or(0)) - } - - /// Reminder rollup stats for an agent over a time window. Returns - /// counts of scheduled, delivered, and pending reminders created - /// in the last `since_secs` seconds (0 = all reminders). - pub fn reminder_rollup_for( - &self, - agent: &str, - since_secs: u64, - ) -> Result { - let conn = self.conn.lock().unwrap(); - let cutoff_time = if since_secs > 0 { - let now = std::time::SystemTime::now() - .duration_since(std::time::UNIX_EPOCH) - .ok() - .and_then(|d| i64::try_from(d.as_secs()).ok()) - .unwrap_or(0); - now.saturating_sub(i64::try_from(since_secs).unwrap_or(i64::MAX)) - } else { - i64::MIN - }; - let scheduled: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2", - params![agent, cutoff_time], - |row| row.get(0), - )?; - let delivered: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NOT NULL", - params![agent, cutoff_time], - |row| row.get(0), - )?; - let pending: i64 = conn.query_row( - "SELECT COUNT(*) FROM reminders WHERE agent = ?1 AND created_at >= ?2 AND sent_at IS NULL", - params![agent, cutoff_time], - |row| row.get(0), - )?; - Ok(hive_sh4re::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), - }) - } - - /// Delete a reminder by id. Returns the number of rows removed (0 - /// when the id never existed or was already delivered). Hard - /// delete rather than soft so the row doesn't linger and confuse a - /// re-creation under the same id. - pub fn cancel_reminder(&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) - } - - /// Cancel a pending reminder on behalf of `canceller`. Returns - /// the owner agent name on success (handy for logging). Auth - /// rules mirror `OperatorQuestions::cancel`: the owner, the - /// operator, or a `privileged` caller (one that arrived on the - /// manager socket — the trust boundary, not a name match). - pub fn cancel_reminder_as(&self, id: i64, canceller: &str, privileged: bool) -> Result { - let conn = self.conn.lock().unwrap(); - let owner: Option = conn - .query_row( - "SELECT agent FROM reminders WHERE id = ?1 AND sent_at IS NULL", - params![id], - |row| row.get(0), - ) - .optional()?; - let Some(owner) = owner else { - anyhow::bail!("reminder {id} not pending (already delivered or unknown)"); - }; - let authorised = - privileged || canceller == owner || canceller == hive_sh4re::OPERATOR_RECIPIENT; - if !authorised { - anyhow::bail!("reminder {id}: '{canceller}' not allowed to cancel (owner = '{owner}')"); - } - let n = conn.execute( - "DELETE FROM reminders WHERE id = ?1 AND sent_at IS NULL", - params![id], - )?; - if n == 0 { - anyhow::bail!("reminder {id} vanished between auth check and delete"); - } - Ok(owner) - } - - /// Get up to `limit` due reminders across all agents in a single query. - /// Returns `(agent, id, message, file_path)` tuples. Pass a small limit - /// (e.g. 100) so a burst of overdue reminders doesn't flood the broker - /// in one cycle — leftovers stay due and get picked up on the next tick. - pub fn get_due_reminders(&self, limit: u64) -> Result> { - let conn = self.conn.lock().unwrap(); - let limit_i = i64::try_from(limit.min(i64::MAX as u64)).unwrap_or(i64::MAX); - let max_attempts = i64::from(MAX_REMINDER_ATTEMPTS); - // attempt_count >= cap = give up; row stays pending so the - // operator sees + can retry/cancel via the dashboard. - let mut stmt = conn.prepare( - "SELECT agent, id, message, file_path FROM reminders \ - WHERE due_at <= ?1 AND sent_at IS NULL AND attempt_count < ?3 \ - ORDER BY agent, due_at ASC \ - LIMIT ?2", - )?; - let rows = stmt.query_map(params![now_unix(), limit_i, max_attempts], |row| { - Ok(( - row.get::<_, String>(0)?, - row.get::<_, i64>(1)?, - row.get::<_, String>(2)?, - row.get::<_, Option>(3)?, - )) - })?; - rows.collect::>>() - .context("query due reminders") - } - - /// Atomic reminder delivery: insert the inbox message AND mark the - /// reminder as sent in a single sqlite transaction. Prevents the - /// orphan-reminder duplicate-delivery class of bugs that two separate - /// calls (send + `mark_reminder_sent`) could produce if the second one - /// failed transiently — the next scheduler tick would see the reminder - /// still due and redeliver. Either both writes commit or neither does; - /// re-running on failure is safe. - /// - /// Emits a `Sent` event on the broadcast channel after the transaction - /// commits (so subscribers see the inbox message but never see a - /// "phantom" send for a transaction that rolled back). - /// Deliver a batch of reminders in a single transaction, reducing - /// lock contention on the shared sqlite connection under high - /// reminder volume. Returns per-item results so the scheduler can - /// record individual failures without aborting successful ones. - /// - /// Items where the INSERT+UPDATE succeeds get a `MessageEvent::Sent` - /// emitted after the transaction commits. Items that fail are - /// returned as `Err` in the output vec (index-aligned with input). - pub fn deliver_reminders_batch( - &self, - items: &[(i64, String, String)], // (reminder_id, agent, body) - ) -> Vec> { - if items.is_empty() { - return Vec::new(); - } - let now = now_unix(); - let mut conn = self.conn.lock().unwrap(); - // Build one transaction for all deliveries so we hold the lock - // once rather than N times. On a batch-level error (e.g. DB - // corruption), fall back to returning per-item errors so the - // scheduler records the failure cleanly. - let tx = match conn.transaction() { - Ok(t) => t, - Err(e) => { - let err_str = format!("{e:#}"); - return items - .iter() - .map(|_| Err(anyhow::anyhow!("{}", err_str.clone()))) - .collect(); - } - }; - let mut results: Vec> = Vec::with_capacity(items.len()); - // Per-item broker row ids — collected inside the transaction so - // we can emit Sent events with the correct id after commit. - let mut msg_ids: Vec = Vec::with_capacity(items.len()); - for (id, agent, body) in items { - let r = (|| -> Result { - tx.execute( - "INSERT INTO messages (sender, recipient, body, sent_at) \ - VALUES (?1, ?2, ?3, ?4)", - params!["reminder", agent, body, now], - )?; - let msg_id = tx.last_insert_rowid(); - tx.execute( - "UPDATE reminders SET sent_at = ?1 WHERE id = ?2", - params![now, id], - )?; - Ok(msg_id) - })(); - match r { - Ok(msg_id) => { - msg_ids.push(msg_id); - results.push(Ok(())); - } - Err(e) => { - msg_ids.push(-1); - results.push(Err(e)); - } - } - } - if let Err(e) = tx.commit() { - let err_str = format!("{e:#}"); - return items - .iter() - .map(|_| Err(anyhow::anyhow!("{}", err_str.clone()))) - .collect(); - } - drop(conn); - // Emit per-row Sent events (only for rows that succeeded). - for (((id, agent, body), result), msg_id) in - items.iter().zip(results.iter()).zip(msg_ids.iter()) - { - if result.is_ok() { - let _ = self.events.send(MessageEvent::Sent { - id: *msg_id, - from: "reminder".to_owned(), - to: agent.clone(), - body: body.clone(), - at: now, - in_reply_to: None, - }); - tracing::debug!(reminder_id = id, %agent, "reminder delivered"); - } - } - results - } } #[cfg(test)] diff --git a/hive-c0re/src/workers/mod.rs b/hive-c0re/src/workers/mod.rs index 75619f0b..ee5fa0ad 100644 --- a/hive-c0re/src/workers/mod.rs +++ b/hive-c0re/src/workers/mod.rs @@ -1,5 +1,5 @@ //! Background tasks and periodic sweeps: crash/login watcher, the -//! reminder and scheduled-prompt delivery loops, boot-time auto-update +//! scheduled-prompt delivery loop, boot-time auto-update //! reconcile, the agent-sockets.json writer loop, the MCP socket listener //! reconcile loop, and knowledge-repo sync. Each submodule is re-exported //! at the crate root, so `crate::crash_watch::…` etc. keep working unchanged. @@ -9,5 +9,4 @@ pub mod auto_update; pub mod crash_watch; pub mod knowledge; pub mod mcp_sockets; -pub mod reminder_scheduler; pub mod scheduled_prompts_worker; diff --git a/hive-c0re/src/workers/reminder_scheduler.rs b/hive-c0re/src/workers/reminder_scheduler.rs deleted file mode 100644 index f05f8d92..00000000 --- a/hive-c0re/src/workers/reminder_scheduler.rs +++ /dev/null @@ -1,278 +0,0 @@ -//! Background loop that drains due reminders from the broker and -//! delivers them as inbox messages. 5s poll cadence, shutdown-aware. -//! File-path semantics (path translation, traversal + symlink defense, -//! pointer delivery): `docs/approvals.md::Reminder delivery`. - -use std::io::Write; -use std::os::unix::fs::OpenOptionsExt; -use std::path::{Path, PathBuf}; -use std::sync::Arc; -use std::time::Duration; - -use crate::coordinator::Coordinator; - -/// Per-tick cap on reminders delivered. Anything over this stays due -/// in the table and gets picked up on the next tick — keeps a -/// 10k-deep backlog from flooding the broker (or hogging the broker -/// mutex) in one shot. 100/tick × 5s tick = sustained throughput cap -/// of ~20 reminders/sec; bump together if the loose-ends tracker -/// starts firing higher rates. -const REMINDER_BATCH_LIMIT: u64 = 100; - -/// Poll interval. Trade-off between latency on a freshly due reminder -/// and CPU spent on empty sweeps; 5s matches the original inline -/// scheduler. -const POLL_INTERVAL: Duration = Duration::from_secs(5); - -pub fn spawn(coord: Arc) { - let mut shutdown = coord.shutdown_rx(); - tokio::spawn(async move { - loop { - tick(&coord); - tokio::select! { - () = tokio::time::sleep(POLL_INTERVAL) => {} - _ = shutdown.changed() => { - tracing::info!("reminder scheduler: shutdown signal received"); - break; - } - } - } - }); -} - -fn tick(coord: &Arc) { - let due = match coord.broker.get_due_reminders(REMINDER_BATCH_LIMIT) { - Ok(rows) => rows, - Err(e) => { - tracing::warn!(error = ?e, "failed to query due reminders"); - return; - } - }; - if due.is_empty() { - return; - } - // Resolve body strings (file-path writes / inline) before entering - // the batch transaction so the DB lock is held as briefly as possible. - let items: Vec<(i64, String, String)> = due - .iter() - .map(|(agent, id, message, file_path)| { - let body = prepare_body(agent, message, file_path.as_deref()); - (*id, agent.clone(), body) - }) - .collect(); - // Single-transaction batch: one DB lock acquisition for N reminders - // instead of N sequential lock/unlock cycles. - let results = coord.broker.deliver_reminders_batch(&items); - let any_delivered = results.iter().any(Result::is_ok); - for ((id, agent, _body), result) in items.iter().zip(results.iter()) { - if let Err(e) = result { - let reason = format!("{e:#}"); - tracing::warn!( - reminder_id = id, - %agent, - error = %reason, - "failed to deliver reminder" - ); - // Persist the failure so the dashboard can surface it. - if let Err(persist_err) = coord.broker.record_reminder_failure(*id, &reason) { - tracing::warn!( - reminder_id = id, - error = ?persist_err, - "failed to persist reminder failure" - ); - } - } - } - // Emit after the batch so the dashboard's pending-reminders list - // updates when deliveries land (removes delivered rows). - if any_delivered { - coord.emit_reminders_snapshot(); - } -} - -/// Build the inbox body for a due reminder. When `file_path` is None -/// the body is the original message verbatim. When set, we attempt to -/// persist the message body to the requested file and return a short -/// pointer string instead. Failures (bad prefix, symlink escape, -/// write error, missing parent) fall back to inline delivery with a -/// noted warning so the reminder still fires. -fn prepare_body(agent: &str, message: &str, file_path: Option<&str>) -> String { - let Some(req_path) = file_path else { - return message.to_owned(); - }; - let host_path = match resolve_host_path(agent, req_path) { - Ok(p) => p, - Err(reason) => { - tracing::warn!(%agent, %req_path, %reason, "reminder file_path rejected; delivering inline"); - return inline_fallback(req_path, &format!("rejected: {reason}"), message); - } - }; - match write_payload(agent, &host_path, message) { - Ok(()) => { - let bytes = message.len(); - // debug! not info! — under load this would dominate the log. - tracing::debug!(%agent, path = %host_path.display(), bytes, "reminder body written to file"); - format!( - "reminder body persisted to `{req_path}` ({bytes} bytes); read with your filesystem tools" - ) - } - Err(reason) => { - tracing::warn!(%agent, path = %host_path.display(), %reason, "reminder file_path write failed; delivering inline"); - inline_fallback(req_path, &reason, message) - } - } -} - -fn inline_fallback(req_path: &str, reason: &str, message: &str) -> String { - format!("[reminder file_path '{req_path}' {reason}; delivering body inline]\n\n{message}") -} - -/// Persist `message` to `host_path` with the symlink-escape defenses -/// described in the module docs. Returns `Ok(())` on success, or a -/// human-readable reason string on any failure (caller logs + -/// inline-falls-back). `pub` because `socket_server::handle_remind` -/// reuses it for the at-remind-time auto-file path. -pub fn write_payload(agent: &str, host_path: &Path, message: &str) -> Result<(), String> { - let agent = hive_types::Ident::parse(agent) - .map_err(|e| format!("invalid agent name {agent:?}: {e}"))?; - let Some(parent) = host_path.parent() else { - return Err("internal: host path has no parent".to_owned()); - }; - std::fs::create_dir_all(parent).map_err(|e| format!("parent dir create failed: {e}"))?; - // Resolve symlinks in the parent chain, then re-verify the - // canonical form still lives under the agent's host state root — - // catches `ln -s /etc state/escape` style attacks. - let parent_canonical = parent - .canonicalize() - .map_err(|e| format!("parent canonicalize failed: {e}"))?; - let agent_root = Coordinator::agent_notes_dir(&agent) - .canonicalize() - .map_err(|e| format!("agent state root canonicalize failed: {e}"))?; - if !parent_canonical.starts_with(&agent_root) { - return Err(format!( - "symlink escape: canonical parent `{}` outside agent root `{}`", - parent_canonical.display(), - agent_root.display() - )); - } - let basename = host_path - .file_name() - .ok_or_else(|| "missing basename".to_owned())?; - let target = parent_canonical.join(basename); - // O_NOFOLLOW on the final component refuses to open if the - // basename is itself an existing symlink. Combined with the - // canonicalize-parent check above, no symlink anywhere in the - // path can redirect the write. - let mut file = std::fs::OpenOptions::new() - .write(true) - .create(true) - .truncate(true) - .custom_flags(libc::O_NOFOLLOW) - .open(&target) - .map_err(|e| format!("open failed: {e}"))?; - file.write_all(message.as_bytes()) - .map_err(|e| format!("write failed: {e}"))?; - Ok(()) -} - -/// Container-visible state prefix the caller's `file_path` must live -/// under. Every agent sees its state at `/agents//state/` -/// (see `lifecycle::set_nspawn_flags`). Auto-file paths use the same -/// prefix so the round-trip is symmetric. -#[must_use] -pub fn container_state_prefix(agent: &str) -> String { - format!("/agents/{agent}/state/") -} - -/// Map an agent-visible container path to the matching host path, -/// validating that it lives under the agent's own state subtree, has -/// a non-empty relative tail, and doesn't try to traverse out via -/// `..`. Returns the host `PathBuf` on success, or a human-readable -/// reason string on rejection. `pub` so `socket_server::handle_remind` -/// can reuse it for the at-remind-time auto-file path. -pub fn resolve_host_path(agent: &str, req_path: &str) -> Result { - let agent = hive_types::Ident::parse(agent) - .map_err(|e| format!("invalid agent name {agent:?}: {e}"))?; - let prefix = container_state_prefix(agent.as_str()); - let Some(rel) = req_path.strip_prefix(&prefix) else { - return Err(format!( - "must be absolute and under `{prefix}` (got `{req_path}`)" - )); - }; - if rel.is_empty() { - return Err("file_path must include a filename, not just the state dir".to_owned()); - } - let rel_path = Path::new(rel); - for comp in rel_path.components() { - match comp { - std::path::Component::Normal(_) => {} - other => { - return Err(format!( - "path component `{other:?}` not allowed (no traversal / absolute / root)" - )); - } - } - } - Ok(Coordinator::agent_notes_dir(&agent).join(rel_path)) -} - -#[cfg(test)] -mod tests { - use super::*; - - #[test] - fn rejects_paths_outside_agent_state() { - assert!(resolve_host_path("foo", "/etc/passwd").is_err()); - assert!(resolve_host_path("foo", "/agents/bar/state/x.md").is_err()); - assert!(resolve_host_path("foo", "relative.md").is_err()); - } - - #[test] - fn rejects_traversal() { - assert!(resolve_host_path("foo", "/agents/foo/state/../../etc/passwd").is_err()); - assert!(resolve_host_path("foo", "/agents/foo/state/./x.md").is_err()); - } - - #[test] - fn rejects_empty_relative_tail() { - // Trailing slash → empty tail. Used to fall through to - // create_dir_all + write-to-dir → confusing inline fallback; - // explicit reject gives a cleaner log. - let err = resolve_host_path("foo", "/agents/foo/state/").unwrap_err(); - assert!(err.contains("must include a filename"), "got: {err}"); - } - - #[test] - fn accepts_well_formed_path() { - let p = resolve_host_path("foo", "/agents/foo/state/reminders/123.md").unwrap(); - assert_eq!( - p, - PathBuf::from("/var/lib/hyperhive/agents/foo/state/reminders/123.md") - ); - } - - #[test] - fn manager_uses_container_name_prefix() { - // Manager's container view of its state is at `/agents/ruth/state/`. - assert_eq!(container_state_prefix("ruth"), "/agents/ruth/state/"); - let p = resolve_host_path("ruth", "/agents/ruth/state/reminders/x.md").unwrap(); - assert_eq!( - p, - PathBuf::from("/var/lib/hyperhive/agents/ruth/state/reminders/x.md") - ); - assert!(resolve_host_path("ruth", "/state/x.md").is_err()); - } - - #[test] - fn prepare_body_passthrough_when_no_file_path() { - let s = prepare_body("foo", "hello world", None); - assert_eq!(s, "hello world"); - } - - #[test] - fn prepare_body_falls_back_inline_on_bad_path() { - let s = prepare_body("foo", "payload", Some("/etc/passwd")); - assert!(s.starts_with("[reminder file_path '/etc/passwd' rejected:")); - assert!(s.contains("payload")); - } -} diff --git a/hive-core-agent-sock/src/lib.rs b/hive-core-agent-sock/src/lib.rs index 90e18c4d..cb390201 100644 --- a/hive-core-agent-sock/src/lib.rs +++ b/hive-core-agent-sock/src/lib.rs @@ -9,7 +9,7 @@ use hive_sh4re::{ CancelLooseEndKind, ContainerInfo, DeliveredMessage, InboxRow, JournalPriority, LooseEnd, - MatrixIdentity, ReminderStats, ReminderTiming, SchedulePromptPayload, WireSchedule, + MatrixIdentity, SchedulePromptPayload, WireSchedule, }; use hive_types::Ident; use serde::{Deserialize, Serialize}; @@ -81,18 +81,6 @@ pub enum Request { /// back via `HelperEvent::QuestionAnswered`: see /// `docs/conventions.md::Question routing (Ask / Answer)`. Answer { id: i64, answer: String }, - /// Schedule a reminder message to be delivered to this agent at a - /// future time. The reminder lands in the agent's inbox as an auto-sent - /// message from `"reminder"`. Use for agent follow-ups (e.g. check task - /// status, retry failed operation). Message length is limited; pass - /// `file_path` to store in a file and get a path-reference message - /// instead. - Remind { - message: String, - timing: ReminderTiming, - #[serde(default)] - file_path: Option, - }, /// Loose-ends view. On the agent socket: `None` = self; direct /// children are always accessible; non-children require the /// `query_agent_state` capability — rejected with an error otherwise; @@ -103,33 +91,6 @@ pub enum Request { #[serde(default, skip_serializing_if = "Option::is_none")] agent: Option, }, - /// Count of pending (un-delivered) reminders. On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - /// Used by the harness's per-turn stats sink. - CountPendingReminders { - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, - /// Reminder statistics: counts of scheduled, delivered, and pending - /// reminders over a time window. `since_secs` filters to reminders - /// created in the last N seconds (0 = all). On the agent socket: - /// same target rules as `GetLooseEnds` (self/children free; - /// non-children require `query_agent_state`; `"*"` rejected). - /// On the manager socket: `None` = self, any name = that agent. - ReminderRollup { - /// Only count reminders created in the last N seconds from now. - /// Pass 0 to include all reminders. - #[serde(default)] - since_secs: u64, - /// Whose reminders to roll up. `None` = the caller's own. - /// `Some("")` = that agent's (requires `query_agent_state` - /// capability on the agent socket; always available on the manager - /// socket). - #[serde(default, skip_serializing_if = "Option::is_none")] - agent: Option, - }, /// Set a free-text status string visible on the dashboard. The harness /// writes `{state_dir}/hyperhive-status` locally before sending this /// request; hive-c0re just triggers a dashboard rescan on receipt. @@ -308,10 +269,6 @@ pub enum Response { /// `GetLooseEnds` result: list of loose ends pending against /// this agent. Ordered newest-first within each kind. LooseEnds { loose_ends: Vec }, - /// `CountPendingReminders` result. - PendingRemindersCount { count: u64 }, - /// `ReminderRollup` result: reminder activity stats for the agent. - ReminderRollup(ReminderStats), /// `GetAgentMeta` result. Per-field semantics + serde defaults /// live in `docs/conventions.md::Agent metadata`. AgentMeta {