From ecd9030305472f46093ddd9ecbdd39e0e3af67aa Mon Sep 17 00:00:00 2001 From: damocles Date: Wed, 22 Jul 2026 21:01:24 +0200 Subject: [PATCH] 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`