wire reminder ops into the in-agent socket dispatch (#2635 inc 1)

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<LooseEnd> },
     /// `CountPendingReminders` result.
     PendingRemindersCount { count: u64 },
+    /// `ReminderRollup` result.
+    ReminderRollup { stats: hive_sh4re::ReminderStats },
     /// Op failed; `message` is operator-facing.
     Err { message: String },
 }

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<S: Surface>(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<S: Surface>(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 {

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<Arc<Reminders>>, tx: mpsc::UnboundedSender<hive_sh4re::DeliveredMessage>) {
+pub async fn run(
+    store: Option<Arc<Reminders>>,
+    tx: mpsc::UnboundedSender<hive_sh4re::DeliveredMessage>,
+) {
     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()

hive-agent/src/reminders.rs

@@ -43,6 +43,7 @@ pub struct Reminder {
     pub message: String,
     pub file_path: Option<String>,
     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<Vec<Reminder>> {
         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<Vec<Reminder>> {
         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<Reminder> {
         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<i64> = {
             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::<rusqlite::Result<Vec<_>>>()

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<PathBuf> {
 /// # Errors
 ///
 /// Returns an error if the socket path is set but can't be bound.
-pub async fn run(store: Arc<Todos>, wake: Arc<Notify>) -> Result<()> {
+pub async fn run(
+    store: Arc<Todos>,
+    wake: Arc<Notify>,
+    reminders: Option<Arc<Reminders>>,
+) -> 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<Todos>, wake: Arc<Notify>) -> 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<UnixListener> {
 
 /// 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::<Request>(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,
     }
 }
 

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<usize> {
+    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`