add reminder delivery timer + main.rs wiring (#2635 inc 1)

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<S: Surface>(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<S: Surface>(socket: &Path, poll_ms: u64) -> Result<()> {
         stats,
         &files,
         todo_wake,
+        reminder_rx,
     )
     .await
 }
@@ -513,6 +536,7 @@ async fn serve_loop<S: Surface>(
     stats: Option<TurnStats>,
     files: &turn::TurnFiles,
     todo_wake: Arc<tokio::sync::Notify>,
+    mut reminder_rx: tokio::sync::mpsc::UnboundedReceiver<hive_sh4re::DeliveredMessage>,
 ) -> Result<()> {
     tracing::info!(socket = %socket.display(), "harness serve");
     S::requeue_inflight(socket).await;
@@ -537,6 +561,7 @@ async fn serve_loop<S: Surface>(
                     biased;
                     o = S::recv_next(socket) => o,
                     () = todo_wake.notified() => RecvOutcome::LocalTodo,
+                    Some(dm) = reminder_rx.recv() => RecvOutcome::Message(dm),
                 }
             } {
                 RecvOutcome::Message(first) => first,

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::<u64>().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<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;
+        return;
+    };
+    loop {
+        tick(&store, &tx);
+        tokio::time::sleep(POLL_INTERVAL).await;
+    }
+}
+
+fn tick(store: &Reminders, tx: &mpsc::UnboundedSender<hive_sh4re::DeliveredMessage>) {
+    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<i64, String> {
+    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>), 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<PathBuf, String> {
+    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<i64> {
+    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"));
+    }
+}